@Test public void testDeclareOutputFields() { TollNotificationBolt bolt = new TollNotificationBolt(); TestDeclarer declarer = new TestDeclarer(); bolt.declareOutputFields(declarer); final int numberOfOutputStreams = 3; Assert.assertEquals(numberOfOutputStreams, declarer.streamIdBuffer.size()); Assert.assertEquals(numberOfOutputStreams, declarer.schemaBuffer.size()); Assert.assertEquals(numberOfOutputStreams, declarer.directBuffer.size()); Fields schema = new Fields(TopologyControl.TYPE_FIELD_NAME, TopologyControl.TIMESTAMP_FIELD_NAME, TopologyControl.EMIT_FIELD_NAME, TopologyControl.VEHICLE_ID_FIELD_NAME, TopologyControl.SPEED_FIELD_NAME, TopologyControl.TOLL_FIELD_NAME); HashMap<String, Fields> expectedStreams = new HashMap<String, Fields>(); expectedStreams.put(TopologyControl.TOLL_NOTIFICATIONS_STREAM_ID, schema); expectedStreams.put(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID, schema); expectedStreams.put(TimestampMerger.FLUSH_STREAM_ID, new Fields("ts")); Assert.assertEquals(expectedStreams.keySet(), new HashSet<String>(declarer.streamIdBuffer)); for(int i = 0; i < numberOfOutputStreams; ++i) { Assert.assertEquals(expectedStreams.get(declarer.streamIdBuffer.get(i)).toList(), declarer.schemaBuffer .get(i).toList()); Assert.assertEquals(new Boolean(false), declarer.directBuffer.get(i)); } }
@Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declareStream(TopologyControl.TOLL_NOTIFICATIONS_STREAM_ID, TollNotification.getSchema()); declarer.declareStream(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID, TollNotification.getSchema()); declarer.declareStream(TimestampMerger.FLUSH_STREAM_ID, new Fields("ts")); }
TollNotificationBolt extends BaseRichBolt { @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declareStream(TopologyControl.TOLL_NOTIFICATIONS_STREAM_ID, TollNotification.getSchema()); declarer.declareStream(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID, TollNotification.getSchema()); declarer.declareStream(TimestampMerger.FLUSH_STREAM_ID, new Fields("ts")); } }
TollNotificationBolt extends BaseRichBolt { @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declareStream(TopologyControl.TOLL_NOTIFICATIONS_STREAM_ID, TollNotification.getSchema()); declarer.declareStream(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID, TollNotification.getSchema()); declarer.declareStream(TimestampMerger.FLUSH_STREAM_ID, new Fields("ts")); }  }
TollNotificationBolt extends BaseRichBolt { @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declareStream(TopologyControl.TOLL_NOTIFICATIONS_STREAM_ID, TollNotification.getSchema()); declarer.declareStream(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID, TollNotification.getSchema()); declarer.declareStream(TimestampMerger.FLUSH_STREAM_ID, new Fields("ts")); }  @Override void prepare(@SuppressWarnings("rawtypes") Map stormConf, TopologyContext context, OutputCollector collector); @Override void execute(Tuple input); @Override void declareOutputFields(OutputFieldsDeclarer declarer); }
TollNotificationBolt extends BaseRichBolt { @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declareStream(TopologyControl.TOLL_NOTIFICATIONS_STREAM_ID, TollNotification.getSchema()); declarer.declareStream(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID, TollNotification.getSchema()); declarer.declareStream(TimestampMerger.FLUSH_STREAM_ID, new Fields("ts")); }  @Override void prepare(@SuppressWarnings("rawtypes") Map stormConf, TopologyContext context, OutputCollector collector); @Override void execute(Tuple input); @Override void declareOutputFields(OutputFieldsDeclarer declarer);  }