@Test public void testExecute() { final int numberOfSegments = 1 + this.r.nextInt(5); final int numberOfMinutes = 5 + this.r.nextInt(5); int numberOfTuples = 0; SegmentIdentifier[] segments = new SegmentIdentifier[numberOfSegments]; for(int i = 0; i < numberOfSegments; ++i) { segments[i] = new SegmentIdentifier(new Integer(this.r.nextInt(2)), new Short( (short)this.r.nextInt(Constants.NUMBER_OF_SEGMENT)), new Short((short)this.r.nextInt(1))); } final LinkedList<Values> expectedResult = new LinkedList<Values>(); final LinkedList<Values> expectedFlushs = new LinkedList<Values>(); Tuple tuple = mock(Tuple.class); when(tuple.getSourceStreamId()).thenReturn("streamId"); OngoingStubbing<List<Object>> tupleStub = when(tuple.getValues()); final int startMinute = 1 + this.r.nextInt(5); for(int m = startMinute; m < startMinute + numberOfMinutes; ++m) { expectedFlushs.add(new Values(new Short((short)((m * 60) - 61)))); final HashMap<SegmentIdentifier, Set<Integer>> counts = new HashMap<SegmentIdentifier, Set<Integer>>(); final int numberOfTuplesMinute = 200 + this.r.nextInt(200); numberOfTuples += numberOfTuplesMinute; for(int i = 0; i < numberOfTuplesMinute; ++i) { int time = (m - 1) * 60; int lane = this.r.nextInt(5); int segIdx = this.r.nextInt(numberOfSegments); Integer vid = new Integer(this.r.nextInt()); tupleStub = tupleStub.thenReturn(new PositionReport(new Short((short)time), vid, new Integer(this.r .nextInt(Constants.NUMBER_OF_SPEEDS)), segments[segIdx].getXWay(), new Short((short)lane), segments[segIdx].getDirection(), segments[segIdx].getSegment(), new Integer(0))); Set<Integer> cnt = counts.get(segments[segIdx]); if(cnt == null) { cnt = new HashSet<Integer>(); cnt.add(vid); counts.put(segments[segIdx], cnt); } else { cnt.add(vid); } } final int dummyIndex = expectedResult.size(); expectedResult.add(new Values(new Short((short)(((m + 1) * 60) - 1)))); for(Entry<SegmentIdentifier, Set<Integer>> e : counts.entrySet()) { SegmentIdentifier segId = e.getKey(); int count = e.getValue().size(); expectedResult.add(new CountTuple(new Short((short)((m * 60) - 1)), segId.getXWay(), segId.getSegment(), segId.getDirection(), new Integer(count))); } if(counts.size() > 0) { expectedResult.remove(dummyIndex); } } tupleStub.thenReturn(null); CountVehiclesBolt bolt = new CountVehiclesBolt(); TestOutputCollector collector = new TestOutputCollector(); bolt.prepare(null, null, new OutputCollector(collector)); for(int i = 0; i < numberOfTuples; ++i) { bolt.execute(tuple); } Tuple flushTuple = mock(Tuple.class); when(flushTuple.getSourceStreamId()).thenReturn(TimestampMerger.FLUSH_STREAM_ID); bolt.execute(flushTuple); expectedFlushs.add(new Values((Object)null)); Assert.assertEquals(2, collector.output.size()); Assert.assertEquals(expectedResult, collector.output.get(TopologyControl.CAR_COUNTS_STREAM_ID)); Assert.assertEquals(expectedFlushs, collector.output.get(TimestampMerger.FLUSH_STREAM_ID)); }
@Override public void execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { if(input.getValue(0) == null) { this.flushBuffer(); this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values((Object)null)); } this.collector.ack(input); return; } this.inputPositionReport.clear(); this.inputPositionReport.addAll(input.getValues()); LOGGER.trace(this.inputPositionReport.toString()); short minute = this.inputPositionReport.getMinuteNumber(); this.segment.set(this.inputPositionReport); assert (minute >= this.currentMinute); if(minute > this.currentMinute) { if(this.countsMap.size() > 0) { this.flushBuffer(); this.countsMap.clear(); } this.currentMinute = minute; this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values(new Short((short)((minute * 60) - 61)))); } Set<Integer> segCnt = this.countsMap.get(this.segment); if(segCnt == null) { segCnt = new HashSet<Integer>(); segCnt.add(this.inputPositionReport.getVid()); this.countsMap.put(this.segment.copy(), segCnt); } else { segCnt.add(this.inputPositionReport.getVid()); } this.collector.ack(input); }
CountVehiclesBolt extends BaseRichBolt { @Override public void execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { if(input.getValue(0) == null) { this.flushBuffer(); this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values((Object)null)); } this.collector.ack(input); return; } this.inputPositionReport.clear(); this.inputPositionReport.addAll(input.getValues()); LOGGER.trace(this.inputPositionReport.toString()); short minute = this.inputPositionReport.getMinuteNumber(); this.segment.set(this.inputPositionReport); assert (minute >= this.currentMinute); if(minute > this.currentMinute) { if(this.countsMap.size() > 0) { this.flushBuffer(); this.countsMap.clear(); } this.currentMinute = minute; this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values(new Short((short)((minute * 60) - 61)))); } Set<Integer> segCnt = this.countsMap.get(this.segment); if(segCnt == null) { segCnt = new HashSet<Integer>(); segCnt.add(this.inputPositionReport.getVid()); this.countsMap.put(this.segment.copy(), segCnt); } else { segCnt.add(this.inputPositionReport.getVid()); } this.collector.ack(input); } }
CountVehiclesBolt extends BaseRichBolt { @Override public void execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { if(input.getValue(0) == null) { this.flushBuffer(); this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values((Object)null)); } this.collector.ack(input); return; } this.inputPositionReport.clear(); this.inputPositionReport.addAll(input.getValues()); LOGGER.trace(this.inputPositionReport.toString()); short minute = this.inputPositionReport.getMinuteNumber(); this.segment.set(this.inputPositionReport); assert (minute >= this.currentMinute); if(minute > this.currentMinute) { if(this.countsMap.size() > 0) { this.flushBuffer(); this.countsMap.clear(); } this.currentMinute = minute; this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values(new Short((short)((minute * 60) - 61)))); } Set<Integer> segCnt = this.countsMap.get(this.segment); if(segCnt == null) { segCnt = new HashSet<Integer>(); segCnt.add(this.inputPositionReport.getVid()); this.countsMap.put(this.segment.copy(), segCnt); } else { segCnt.add(this.inputPositionReport.getVid()); } this.collector.ack(input); }  }
CountVehiclesBolt extends BaseRichBolt { @Override public void execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { if(input.getValue(0) == null) { this.flushBuffer(); this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values((Object)null)); } this.collector.ack(input); return; } this.inputPositionReport.clear(); this.inputPositionReport.addAll(input.getValues()); LOGGER.trace(this.inputPositionReport.toString()); short minute = this.inputPositionReport.getMinuteNumber(); this.segment.set(this.inputPositionReport); assert (minute >= this.currentMinute); if(minute > this.currentMinute) { if(this.countsMap.size() > 0) { this.flushBuffer(); this.countsMap.clear(); } this.currentMinute = minute; this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values(new Short((short)((minute * 60) - 61)))); } Set<Integer> segCnt = this.countsMap.get(this.segment); if(segCnt == null) { segCnt = new HashSet<Integer>(); segCnt.add(this.inputPositionReport.getVid()); this.countsMap.put(this.segment.copy(), segCnt); } else { segCnt.add(this.inputPositionReport.getVid()); } this.collector.ack(input); }  @Override void prepare(@SuppressWarnings("rawtypes") Map stormConf, TopologyContext context, OutputCollector collector); @Override void execute(Tuple input); @Override void declareOutputFields(OutputFieldsDeclarer declarer); }
CountVehiclesBolt extends BaseRichBolt { @Override public void execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { if(input.getValue(0) == null) { this.flushBuffer(); this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values((Object)null)); } this.collector.ack(input); return; } this.inputPositionReport.clear(); this.inputPositionReport.addAll(input.getValues()); LOGGER.trace(this.inputPositionReport.toString()); short minute = this.inputPositionReport.getMinuteNumber(); this.segment.set(this.inputPositionReport); assert (minute >= this.currentMinute); if(minute > this.currentMinute) { if(this.countsMap.size() > 0) { this.flushBuffer(); this.countsMap.clear(); } this.currentMinute = minute; this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values(new Short((short)((minute * 60) - 61)))); } Set<Integer> segCnt = this.countsMap.get(this.segment); if(segCnt == null) { segCnt = new HashSet<Integer>(); segCnt.add(this.inputPositionReport.getVid()); this.countsMap.put(this.segment.copy(), segCnt); } else { segCnt.add(this.inputPositionReport.getVid()); } this.collector.ack(input); }  @Override void prepare(@SuppressWarnings("rawtypes") Map stormConf, TopologyContext context, OutputCollector collector); @Override void execute(Tuple input); @Override void declareOutputFields(OutputFieldsDeclarer declarer);  }