Let's wire in the JSON bolt to parse this JSON payload in our topology and see the execution:
...
public class JsonBolt implements IRichBolt {
..
private static final long serialVersionUID = 1L;
OutputCollector collector;
public void prepare(Map arg0, TopologyContext arg1, OutputCollector arg2) {
this.collector = arg2;
}
public void execute(Tuple arg0) {
String jsonInString = arg0.getStringByField("messages");
System.out.println("message read from queue" + jsonInString);
JsonConverter jsonconvertor = new JsonConverter();
MySensorData mysensorObj = jsonconvertor.run(jsonInString);
this.collector.emit(new Values(mysensorObj));
this.collector.ack(arg0);
}
..
}
..
class JsonConverter {
..
public MySensorData run(String jsonInString) {
ObjectMapper mapper = new ObjectMapper...