JavaToSiddhiSerializer implements Serializer<StratioStreamingMessage, Event> { @Override public Event serialize(StratioStreamingMessage object) { Object[] values = new Object[object.getColumns().size()]; for (ColumnNameTypeValue column : object.getColumns()) { values[streamMetadataService.getAttributePosition(object.getStreamName(), column.getColumn())] = column .getValue(); } return new InEvent(object.getStreamName(), System.currentTimeMillis(), values); } JavaToSiddhiSerializer(StreamMetadataService streamMetadataService); @Override Event serialize(StratioStreamingMessage object); @Override StratioStreamingMessage deserialize(Event object); @Override List<Event> serialize(List<StratioStreamingMessage> object); @Override List<StratioStreamingMessage> deserialize(List<Event> object);  }
@Test public void testSerialize() throws Exception { StratioStreamingMessage message= StreamsHelper.getSampleMessage(); Event event= serializer.serialize(message); assertEquals("Expected value not found", "name", event.getData(0)); assertEquals("Expected value not found", 3, event.getData(2)); assertEquals("Expected value not found", 13f, event.getData(5)); }
@Test public void testSerializeList() throws Exception { List<StratioStreamingMessage> messages= new ArrayList<>(); messages.add(StreamsHelper.getSampleMessage()); messages.add(StreamsHelper.getSampleMessage()); List<Event> events= serializer.serialize(messages); assertEquals("Expected value not found", 2, events.size()); }