@SuppressWarnings("boxing") @Test public void testExecute() { List<Values> expectedResult = new ArrayList<Values>(); expectedResult.add(new PositionReport((short)8400, 32, 0, 0, (short)1, (short)0, (short)2, 11559)); expectedResult.add(new AccountBalanceRequest((short)8410, 106483, 42)); expectedResult.add(new DailyExpenditureRequest((short)8420, 104954, 4, 43, (short)3)); expectedResult.add(new TravelTimeRequest((short)8430, 104596, 3, 44, (short)5, (short)8, (short)4, (short)67)); Tuple tuple = mock(Tuple.class); when(tuple.getSourceStreamId()).thenReturn("streamId"); OngoingStubbing<String> tupleStub1 = when(tuple.getString(1)); tupleStub1 = tupleStub1.thenReturn("0,8400,32,0,0,1,0,2,11559,-1,-1,-1,-1,-1,-1"); tupleStub1 = tupleStub1.thenReturn("2,8410,106483,-1,-1,-1,-1,-1,-1,42,-1,-1,-1,-1,-1"); tupleStub1 = tupleStub1.thenReturn("3,8420,104954,-1,4,-1,v,-1,-1,43,-1,-1,-1,-1,3"); tupleStub1 = tupleStub1.thenReturn("4,8430,104596,-1,3,-1,-1,-1,-1,44,5,8,4,67,-1"); tupleStub1.thenReturn(null); OngoingStubbing<Long> tupleStub2 = when(tuple.getLong(0)); tupleStub2 = tupleStub2.thenReturn(8400L); tupleStub2 = tupleStub2.thenReturn(8410L); tupleStub2 = tupleStub2.thenReturn(8420L); tupleStub2 = tupleStub2.thenReturn(8430L); tupleStub2.thenReturn(null); DispatcherBolt bolt = new DispatcherBolt(); TestOutputCollector collector = new TestOutputCollector(); bolt.prepare(null, null, new OutputCollector(collector)); for(int i = 0; i < 4; ++i) { bolt.execute(tuple); } Assert.assertEquals(4, collector.output.size()); Assert.assertEquals(1, collector.output.get(TopologyControl.POSITION_REPORTS_STREAM_ID).size()); Assert.assertEquals(1, collector.output.get(TopologyControl.ACCOUNT_BALANCE_REQUESTS_STREAM_ID).size()); Assert.assertEquals(1, collector.output.get(TopologyControl.DAILY_EXPEDITURE_REQUESTS_STREAM_ID).size()); Assert.assertEquals(1, collector.output.get(TopologyControl.TRAVEL_TIME_REQUEST_STREAM_ID).size()); Assert.assertEquals(expectedResult.get(0), collector.output.get(TopologyControl.POSITION_REPORTS_STREAM_ID) .get(0)); Assert.assertEquals(expectedResult.get(1), collector.output.get(TopologyControl.ACCOUNT_BALANCE_REQUESTS_STREAM_ID).get(0)); Assert.assertEquals(expectedResult.get(2), collector.output.get(TopologyControl.DAILY_EXPEDITURE_REQUESTS_STREAM_ID).get(0)); Assert.assertEquals(expectedResult.get(3), collector.output.get(TopologyControl.TRAVEL_TIME_REQUEST_STREAM_ID) .get(0)); Assert.assertNull(collector.output.get(TimestampMerger.FLUSH_STREAM_ID)); Tuple flushTuple = mock(Tuple.class); when(flushTuple.getSourceStreamId()).thenReturn(TimestampMerger.FLUSH_STREAM_ID); bolt.execute(flushTuple); Assert.assertEquals(1, collector.output.get(TimestampMerger.FLUSH_STREAM_ID).size()); Assert.assertEquals(new Values((Object)null), collector.output.get(TimestampMerger.FLUSH_STREAM_ID).get(0)); }
@Override public void execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values((Object)null)); this.collector.ack(input); return; } String raw = null; try { raw = input.getString(1); String[] token = raw.split(","); short type = Short.parseShort(token[0]); Short time = new Short(input.getLong(0).shortValue()); Integer vid = new Integer(Integer.parseInt(token[2])); assert (time.shortValue() == Short.parseShort(token[1])); if(type == AbstractLRBTuple.position_report) { this.collector.emit(TopologyControl.POSITION_REPORTS_STREAM_ID, new PositionReport( time, vid, new Integer(Integer.parseInt(token[3])), new Integer(Integer.parseInt(token[4])), new Short(Short.parseShort(token[5])), new Short(Short.parseShort(token[6])), new Short(Short.parseShort(token[7])), new Integer(Integer.parseInt(token[8])))); } else { Integer qid = new Integer(Integer.parseInt(token[9])); switch(type) { case AbstractLRBTuple.account_balance_request: this.collector.emit(TopologyControl.ACCOUNT_BALANCE_REQUESTS_STREAM_ID, new AccountBalanceRequest( time, vid, qid)); break; case AbstractLRBTuple.daily_expenditure_request: this.collector.emit(TopologyControl.DAILY_EXPEDITURE_REQUESTS_STREAM_ID, new DailyExpenditureRequest(time, vid, new Integer(Integer.parseInt(token[4])), qid, new Short(Short.parseShort(token[14])))); break; case AbstractLRBTuple.travel_time_request: this.collector.emit(TopologyControl.TRAVEL_TIME_REQUEST_STREAM_ID, new TravelTimeRequest(time, vid, new Integer(Integer.parseInt(token[4])), qid, new Short(Short.parseShort(token[10])), new Short(Short.parseShort(token[11])), new Short(Short.parseShort(token[12])), new Short(Short.parseShort(token[13])))); break; default: LOGGER.error("Unkown tuple type: {}", new Short(type)); } } } catch(Exception e) { LOGGER.error("Error in line: {}", raw); LOGGER.error("StackTrace:", e); } this.collector.ack(input); }
DispatcherBolt extends BaseRichBolt { @Override public void execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values((Object)null)); this.collector.ack(input); return; } String raw = null; try { raw = input.getString(1); String[] token = raw.split(","); short type = Short.parseShort(token[0]); Short time = new Short(input.getLong(0).shortValue()); Integer vid = new Integer(Integer.parseInt(token[2])); assert (time.shortValue() == Short.parseShort(token[1])); if(type == AbstractLRBTuple.position_report) { this.collector.emit(TopologyControl.POSITION_REPORTS_STREAM_ID, new PositionReport( time, vid, new Integer(Integer.parseInt(token[3])), new Integer(Integer.parseInt(token[4])), new Short(Short.parseShort(token[5])), new Short(Short.parseShort(token[6])), new Short(Short.parseShort(token[7])), new Integer(Integer.parseInt(token[8])))); } else { Integer qid = new Integer(Integer.parseInt(token[9])); switch(type) { case AbstractLRBTuple.account_balance_request: this.collector.emit(TopologyControl.ACCOUNT_BALANCE_REQUESTS_STREAM_ID, new AccountBalanceRequest( time, vid, qid)); break; case AbstractLRBTuple.daily_expenditure_request: this.collector.emit(TopologyControl.DAILY_EXPEDITURE_REQUESTS_STREAM_ID, new DailyExpenditureRequest(time, vid, new Integer(Integer.parseInt(token[4])), qid, new Short(Short.parseShort(token[14])))); break; case AbstractLRBTuple.travel_time_request: this.collector.emit(TopologyControl.TRAVEL_TIME_REQUEST_STREAM_ID, new TravelTimeRequest(time, vid, new Integer(Integer.parseInt(token[4])), qid, new Short(Short.parseShort(token[10])), new Short(Short.parseShort(token[11])), new Short(Short.parseShort(token[12])), new Short(Short.parseShort(token[13])))); break; default: LOGGER.error("Unkown tuple type: {}", new Short(type)); } } } catch(Exception e) { LOGGER.error("Error in line: {}", raw); LOGGER.error("StackTrace:", e); } this.collector.ack(input); } }
DispatcherBolt extends BaseRichBolt { @Override public void execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values((Object)null)); this.collector.ack(input); return; } String raw = null; try { raw = input.getString(1); String[] token = raw.split(","); short type = Short.parseShort(token[0]); Short time = new Short(input.getLong(0).shortValue()); Integer vid = new Integer(Integer.parseInt(token[2])); assert (time.shortValue() == Short.parseShort(token[1])); if(type == AbstractLRBTuple.position_report) { this.collector.emit(TopologyControl.POSITION_REPORTS_STREAM_ID, new PositionReport( time, vid, new Integer(Integer.parseInt(token[3])), new Integer(Integer.parseInt(token[4])), new Short(Short.parseShort(token[5])), new Short(Short.parseShort(token[6])), new Short(Short.parseShort(token[7])), new Integer(Integer.parseInt(token[8])))); } else { Integer qid = new Integer(Integer.parseInt(token[9])); switch(type) { case AbstractLRBTuple.account_balance_request: this.collector.emit(TopologyControl.ACCOUNT_BALANCE_REQUESTS_STREAM_ID, new AccountBalanceRequest( time, vid, qid)); break; case AbstractLRBTuple.daily_expenditure_request: this.collector.emit(TopologyControl.DAILY_EXPEDITURE_REQUESTS_STREAM_ID, new DailyExpenditureRequest(time, vid, new Integer(Integer.parseInt(token[4])), qid, new Short(Short.parseShort(token[14])))); break; case AbstractLRBTuple.travel_time_request: this.collector.emit(TopologyControl.TRAVEL_TIME_REQUEST_STREAM_ID, new TravelTimeRequest(time, vid, new Integer(Integer.parseInt(token[4])), qid, new Short(Short.parseShort(token[10])), new Short(Short.parseShort(token[11])), new Short(Short.parseShort(token[12])), new Short(Short.parseShort(token[13])))); break; default: LOGGER.error("Unkown tuple type: {}", new Short(type)); } } } catch(Exception e) { LOGGER.error("Error in line: {}", raw); LOGGER.error("StackTrace:", e); } this.collector.ack(input); }  }
DispatcherBolt extends BaseRichBolt { @Override public void execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values((Object)null)); this.collector.ack(input); return; } String raw = null; try { raw = input.getString(1); String[] token = raw.split(","); short type = Short.parseShort(token[0]); Short time = new Short(input.getLong(0).shortValue()); Integer vid = new Integer(Integer.parseInt(token[2])); assert (time.shortValue() == Short.parseShort(token[1])); if(type == AbstractLRBTuple.position_report) { this.collector.emit(TopologyControl.POSITION_REPORTS_STREAM_ID, new PositionReport( time, vid, new Integer(Integer.parseInt(token[3])), new Integer(Integer.parseInt(token[4])), new Short(Short.parseShort(token[5])), new Short(Short.parseShort(token[6])), new Short(Short.parseShort(token[7])), new Integer(Integer.parseInt(token[8])))); } else { Integer qid = new Integer(Integer.parseInt(token[9])); switch(type) { case AbstractLRBTuple.account_balance_request: this.collector.emit(TopologyControl.ACCOUNT_BALANCE_REQUESTS_STREAM_ID, new AccountBalanceRequest( time, vid, qid)); break; case AbstractLRBTuple.daily_expenditure_request: this.collector.emit(TopologyControl.DAILY_EXPEDITURE_REQUESTS_STREAM_ID, new DailyExpenditureRequest(time, vid, new Integer(Integer.parseInt(token[4])), qid, new Short(Short.parseShort(token[14])))); break; case AbstractLRBTuple.travel_time_request: this.collector.emit(TopologyControl.TRAVEL_TIME_REQUEST_STREAM_ID, new TravelTimeRequest(time, vid, new Integer(Integer.parseInt(token[4])), qid, new Short(Short.parseShort(token[10])), new Short(Short.parseShort(token[11])), new Short(Short.parseShort(token[12])), new Short(Short.parseShort(token[13])))); break; default: LOGGER.error("Unkown tuple type: {}", new Short(type)); } } } catch(Exception e) { LOGGER.error("Error in line: {}", raw); LOGGER.error("StackTrace:", e); } this.collector.ack(input); }  @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 execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values((Object)null)); this.collector.ack(input); return; } String raw = null; try { raw = input.getString(1); String[] token = raw.split(","); short type = Short.parseShort(token[0]); Short time = new Short(input.getLong(0).shortValue()); Integer vid = new Integer(Integer.parseInt(token[2])); assert (time.shortValue() == Short.parseShort(token[1])); if(type == AbstractLRBTuple.position_report) { this.collector.emit(TopologyControl.POSITION_REPORTS_STREAM_ID, new PositionReport( time, vid, new Integer(Integer.parseInt(token[3])), new Integer(Integer.parseInt(token[4])), new Short(Short.parseShort(token[5])), new Short(Short.parseShort(token[6])), new Short(Short.parseShort(token[7])), new Integer(Integer.parseInt(token[8])))); } else { Integer qid = new Integer(Integer.parseInt(token[9])); switch(type) { case AbstractLRBTuple.account_balance_request: this.collector.emit(TopologyControl.ACCOUNT_BALANCE_REQUESTS_STREAM_ID, new AccountBalanceRequest( time, vid, qid)); break; case AbstractLRBTuple.daily_expenditure_request: this.collector.emit(TopologyControl.DAILY_EXPEDITURE_REQUESTS_STREAM_ID, new DailyExpenditureRequest(time, vid, new Integer(Integer.parseInt(token[4])), qid, new Short(Short.parseShort(token[14])))); break; case AbstractLRBTuple.travel_time_request: this.collector.emit(TopologyControl.TRAVEL_TIME_REQUEST_STREAM_ID, new TravelTimeRequest(time, vid, new Integer(Integer.parseInt(token[4])), qid, new Short(Short.parseShort(token[10])), new Short(Short.parseShort(token[11])), new Short(Short.parseShort(token[12])), new Short(Short.parseShort(token[13])))); break; default: LOGGER.error("Unkown tuple type: {}", new Short(type)); } } } catch(Exception e) { LOGGER.error("Error in line: {}", raw); LOGGER.error("StackTrace:", e); } this.collector.ack(input); }  @Override void prepare(@SuppressWarnings("rawtypes") Map conf, TopologyContext topologyContext, OutputCollector outputCollector); @Override void execute(Tuple input); @Override void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer);  }