StreamToActionBusCallback extends ActionControllerCallback { @Override public void receive(long timeStamp, Event[] inEvents, Event[] removeEvents) { if (log.isDebugEnabled()) { log.debug("Receiving {} events from stream {}", inEvents.length, streamName); } String topicAction = InternalTopic.TOPIC_ACTION.getTopicName(); if (groupId !=null){ topicAction = topicAction.concat("_").concat(groupId); } List<KeyedMessage<String, byte[]>> messages = new ArrayList<>(); for (Event event : inEvents) { StratioStreamingMessage messageObject = javaToSiddhiSerializer.deserialize(event); messageObject.setStreamName(streamName); messageObject.setActiveActions(this.activeActions); messages.add(new KeyedMessage<String, byte[]>(topicAction, javaToAvroSerializer.serialize(messageObject))); } if (messages.size() != 0) { avroProducer.send(messages); } } StreamToActionBusCallback(Set<StreamAction> activeActions, String streamName,
            Producer<String, byte[]> avroProducer,
            Serializer<StratioStreamingMessage, Event> javaToSiddhiSerializer,
            Serializer<StratioStreamingMessage, byte[]> javaToAvroSerializer);  StreamToActionBusCallback(Set<StreamAction> activeActions, String streamName,
            Producer<String, byte[]> avroProducer,
            Serializer<StratioStreamingMessage, Event> javaToSiddhiSerializer,
            Serializer<StratioStreamingMessage, byte[]> javaToAvroSerializer,
            String groupId); @Override void receive(long timeStamp, Event[] inEvents, Event[] removeEvents);  }
@Test public void testReceive() throws Exception { StratioStreamingMessage message= StreamsHelper.getSampleMessage(); Event event= javaToSiddhiSerializer.serialize(message); Event[] inEvents= {event}; Event[] removeEvents= {event}; Exception ex= null; try { cbk.receive(123456789L, inEvents, removeEvents); } catch (Exception e) {ex= e;} assertNull("Unexpected exception found", ex); }