KafkaToJavaSerializer implements Serializer<String, StratioStreamingMessage> { @Override public String deserialize(StratioStreamingMessage object) { return gson.toJson(object); } KafkaToJavaSerializer(Gson gson); @Override StratioStreamingMessage serialize(String object); @Override String deserialize(StratioStreamingMessage object); @Override List<StratioStreamingMessage> serialize(List<String> object); @Override List<String> deserialize(List<StratioStreamingMessage> object);  }
@Test public void testDeserialize() throws Exception { StratioStreamingMessage message= StreamsHelper.getSampleMessage(); String result= serializer.deserialize(message); Gson gson = new Gson(); StratioStreamingMessage des= gson.fromJson(result, StratioStreamingMessage.class); assertEquals("Expected stream name not found", message.getStreamName(), des.getStreamName()); assertEquals("Expected size not found", message.getColumns().size(), des.getColumns().size()); }
@Test public void testDeserializeList() throws Exception { List<StratioStreamingMessage> messages= new ArrayList<StratioStreamingMessage>(); messages.add(StreamsHelper.getSampleMessage()); messages.add(StreamsHelper.getSampleMessage()); List<String> results= serializer.deserialize(messages); assertEquals("Expected size not found", 2, results.size()); }