@Test public void testDeclareOutputFields() { DispatcherBolt bolt = new DispatcherBolt(); TestDeclarer declarer = new TestDeclarer(); bolt.declareOutputFields(declarer); final int numberOfOutputStreams = 5; Assert.assertEquals(numberOfOutputStreams, declarer.streamIdBuffer.size()); Assert.assertEquals(numberOfOutputStreams, declarer.schemaBuffer.size()); Assert.assertEquals(numberOfOutputStreams, declarer.directBuffer.size()); HashMap<String, Fields> expectedStreams = new HashMap<String, Fields>(); expectedStreams.put(TopologyControl.POSITION_REPORTS_STREAM_ID, new Fields(TopologyControl.TYPE_FIELD_NAME, TopologyControl.TIMESTAMP_FIELD_NAME, TopologyControl.VEHICLE_ID_FIELD_NAME, TopologyControl.SPEED_FIELD_NAME, TopologyControl.XWAY_FIELD_NAME, TopologyControl.LANE_FIELD_NAME, TopologyControl.DIRECTION_FIELD_NAME, TopologyControl.SEGMENT_FIELD_NAME, TopologyControl.POSITION_FIELD_NAME)); expectedStreams.put(TopologyControl.ACCOUNT_BALANCE_REQUESTS_STREAM_ID, new Fields( TopologyControl.TYPE_FIELD_NAME, TopologyControl.TIMESTAMP_FIELD_NAME, TopologyControl.VEHICLE_ID_FIELD_NAME, TopologyControl.QUERY_ID_FIELD_NAME)); expectedStreams.put(TopologyControl.DAILY_EXPEDITURE_REQUESTS_STREAM_ID, new Fields( TopologyControl.TYPE_FIELD_NAME, TopologyControl.TIMESTAMP_FIELD_NAME, TopologyControl.VEHICLE_ID_FIELD_NAME, TopologyControl.XWAY_FIELD_NAME, TopologyControl.QUERY_ID_FIELD_NAME, TopologyControl.DAY_FIELD_NAME)); expectedStreams.put(TopologyControl.TRAVEL_TIME_REQUEST_STREAM_ID, new Fields(TopologyControl.TYPE_FIELD_NAME, TopologyControl.TIMESTAMP_FIELD_NAME, TopologyControl.VEHICLE_ID_FIELD_NAME, TopologyControl.XWAY_FIELD_NAME, TopologyControl.QUERY_ID_FIELD_NAME, TopologyControl.START_SEGMENT_FIELD_NAME, TopologyControl.END_SEGMENT_FIELD_NAME, TopologyControl.DAY_OF_WEEK_FIELD_NAME, TopologyControl.TIME_OF_DAY_FIELD_NAME)); 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 outputFieldsDeclarer) { outputFieldsDeclarer.declareStream(TopologyControl.POSITION_REPORTS_STREAM_ID, PositionReport.getSchema()); outputFieldsDeclarer.declareStream(TopologyControl.ACCOUNT_BALANCE_REQUESTS_STREAM_ID, AccountBalanceRequest.getSchema()); outputFieldsDeclarer.declareStream(TopologyControl.DAILY_EXPEDITURE_REQUESTS_STREAM_ID, DailyExpenditureRequest.getSchema()); outputFieldsDeclarer .declareStream(TopologyControl.TRAVEL_TIME_REQUEST_STREAM_ID, TravelTimeRequest.getSchema()); outputFieldsDeclarer.declareStream(TimestampMerger.FLUSH_STREAM_ID, new Fields("ts")); }
DispatcherBolt extends BaseRichBolt { @Override public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) { outputFieldsDeclarer.declareStream(TopologyControl.POSITION_REPORTS_STREAM_ID, PositionReport.getSchema()); outputFieldsDeclarer.declareStream(TopologyControl.ACCOUNT_BALANCE_REQUESTS_STREAM_ID, AccountBalanceRequest.getSchema()); outputFieldsDeclarer.declareStream(TopologyControl.DAILY_EXPEDITURE_REQUESTS_STREAM_ID, DailyExpenditureRequest.getSchema()); outputFieldsDeclarer .declareStream(TopologyControl.TRAVEL_TIME_REQUEST_STREAM_ID, TravelTimeRequest.getSchema()); outputFieldsDeclarer.declareStream(TimestampMerger.FLUSH_STREAM_ID, new Fields("ts")); } }
DispatcherBolt extends BaseRichBolt { @Override public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) { outputFieldsDeclarer.declareStream(TopologyControl.POSITION_REPORTS_STREAM_ID, PositionReport.getSchema()); outputFieldsDeclarer.declareStream(TopologyControl.ACCOUNT_BALANCE_REQUESTS_STREAM_ID, AccountBalanceRequest.getSchema()); outputFieldsDeclarer.declareStream(TopologyControl.DAILY_EXPEDITURE_REQUESTS_STREAM_ID, DailyExpenditureRequest.getSchema()); outputFieldsDeclarer .declareStream(TopologyControl.TRAVEL_TIME_REQUEST_STREAM_ID, TravelTimeRequest.getSchema()); outputFieldsDeclarer.declareStream(TimestampMerger.FLUSH_STREAM_ID, new Fields("ts")); }  }
DispatcherBolt extends BaseRichBolt { @Override public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) { outputFieldsDeclarer.declareStream(TopologyControl.POSITION_REPORTS_STREAM_ID, PositionReport.getSchema()); outputFieldsDeclarer.declareStream(TopologyControl.ACCOUNT_BALANCE_REQUESTS_STREAM_ID, AccountBalanceRequest.getSchema()); outputFieldsDeclarer.declareStream(TopologyControl.DAILY_EXPEDITURE_REQUESTS_STREAM_ID, DailyExpenditureRequest.getSchema()); outputFieldsDeclarer .declareStream(TopologyControl.TRAVEL_TIME_REQUEST_STREAM_ID, TravelTimeRequest.getSchema()); outputFieldsDeclarer.declareStream(TimestampMerger.FLUSH_STREAM_ID, new Fields("ts")); }  @Override void prepare(@SuppressWarnings("rawtypes") Map conf, TopologyContext topologyContext, OutputCollector outputCollector); @Override void execute(Tuple input); @Override void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer); }
DispatcherBolt extends BaseRichBolt { @Override public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) { outputFieldsDeclarer.declareStream(TopologyControl.POSITION_REPORTS_STREAM_ID, PositionReport.getSchema()); outputFieldsDeclarer.declareStream(TopologyControl.ACCOUNT_BALANCE_REQUESTS_STREAM_ID, AccountBalanceRequest.getSchema()); outputFieldsDeclarer.declareStream(TopologyControl.DAILY_EXPEDITURE_REQUESTS_STREAM_ID, DailyExpenditureRequest.getSchema()); outputFieldsDeclarer .declareStream(TopologyControl.TRAVEL_TIME_REQUEST_STREAM_ID, TravelTimeRequest.getSchema()); outputFieldsDeclarer.declareStream(TimestampMerger.FLUSH_STREAM_ID, new Fields("ts")); }  @Override void prepare(@SuppressWarnings("rawtypes") Map conf, TopologyContext topologyContext, OutputCollector outputCollector); @Override void execute(Tuple input); @Override void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer);  }