JavaToSiddhiSerializer implements Serializer<StratioStreamingMessage, Event> { @Override public StratioStreamingMessage deserialize(Event object) { List<ColumnNameTypeValue> columns = new ArrayList<>(); for (int i = 0; i < object.getData().length; i++) { Object data = object.getData()[i]; Attribute attribute = streamMetadataService.getAttribute(object.getStreamId(), i); columns.add(new ColumnNameTypeValue(attribute.getName(), encodeSiddhiType(attribute.getType()), data)); } return new StratioStreamingMessage(object.getStreamId(), object.getTimeStamp(), columns); } 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 testDeserialize() throws Exception { Object[] values = new Object[5]; values[0]= new ColumnNameTypeValue("name", ColumnType.STRING, "test"); values[1]= new ColumnNameTypeValue("value", ColumnType.INTEGER, 10); values[2]= new ColumnNameTypeValue("enabled", ColumnType.BOOLEAN, true); values[3]= new ColumnNameTypeValue("numberl", ColumnType.LONG, 123L); values[4]= new ColumnNameTypeValue("numberf", ColumnType.FLOAT, 13f); Event event= new InEvent(StreamsHelper.STREAM_NAME, System.currentTimeMillis(), values); StratioStreamingMessage message= serializer.deserialize(event); assertEquals("Expected value not found", "test", ((ColumnNameTypeValue) message.getColumns().get(0).getValue()).getValue()); assertEquals("Expected value not found",10, ((ColumnNameTypeValue) message.getColumns().get(1).getValue()).getValue()); assertEquals("Expected value not found", true, ((ColumnNameTypeValue) message.getColumns().get(2).getValue()).getValue()); }
@Test public void testDeserializeList() throws Exception { List<Event> events= new ArrayList<Event>(); Object[] values = new Object[3]; values[0]= new ColumnNameTypeValue("name", ColumnType.STRING, "test"); values[1]= new ColumnNameTypeValue("value", ColumnType.INTEGER, 10); values[2]= new ColumnNameTypeValue("enabled", ColumnType.BOOLEAN, true); events.add(new InEvent(StreamsHelper.STREAM_NAME, System.currentTimeMillis(), values)); values = new Object[3]; values[0]= new ColumnNameTypeValue("name", ColumnType.STRING, "test2"); values[1]= new ColumnNameTypeValue("value", ColumnType.INTEGER, 11); values[2]= new ColumnNameTypeValue("enabled", ColumnType.BOOLEAN, false); events.add(new InEvent(StreamsHelper.STREAM_NAME, System.currentTimeMillis(), values)); List<StratioStreamingMessage> messages= serializer.deserialize(events); assertEquals("Expected value not found", 2, messages.size()); }