@SuppressWarnings("boxing") @Test public void testExecute() { Tuple tuple = mock(Tuple.class); when(tuple.getSourceStreamId()).thenReturn("streamId"); OngoingStubbing<List<Object>> tupleValueStub = when(tuple.getValues()); final int numberOfVehicles = timeAndPositions.length; final int numberOfReporst = timeAndPositions[0].length; for(int m = 0; m < numberOfReporst; ++m) { for(int vid = 1; vid <= numberOfVehicles; ++vid) { PositionReport pr = new PositionReport(timeAndPositions[vid - 1][m][1].shortValue(), vid, 0, 0, timeAndPositions[vid - 1][m][0].shortValue(), Constants.EASTBOUND, (short)(timeAndPositions[vid - 1][m][2].intValue() / 5280), timeAndPositions[vid - 1][m][2]); tupleValueStub = tupleValueStub.thenReturn(pr); } } tupleValueStub.thenReturn(null); StoppedCarsBolt bolt = new StoppedCarsBolt(); TestOutputCollector collector = new TestOutputCollector(); bolt.prepare(null, null, new OutputCollector(collector)); for(int i = 0; i < numberOfReporst * numberOfVehicles; ++i) { bolt.execute(tuple); } List<Values> expectedFlushs = new ArrayList<Values>(); expectedFlushs.add(new Values((short)1)); expectedFlushs.add(new Values((short)61)); expectedFlushs.add(new Values((short)121)); expectedFlushs.add(new Values((short)181)); expectedFlushs.add(new Values((short)241)); expectedFlushs.add(new Values((short)301)); expectedFlushs.add(new Values((short)361)); List<StoppedCarTuple> expectedResult = new ArrayList<StoppedCarTuple>(); expectedResult.add(new StoppedCarTuple(3, (short)100, 0, (short)0, 50105, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(1, (short)121, 0, (short)0, 50102, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(3, (short)130, 0, (short)0, 50105, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(-1, (short)151, 0, (short)0, 50102, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(2, (short)155, 0, (short)0, 50102, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(3, (short)160, 0, (short)0, 50105, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(-2, (short)185, 0, (short)0, 50102, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(3, (short)190, 0, (short)0, 50105, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(3, (short)220, 0, (short)0, 50105, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(1, (short)241, 0, (short)0, 50103, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(3, (short)250, 0, (short)0, 50105, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(1, (short)271, 0, (short)0, 50103, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(3, (short)280, 0, (short)0, 50105, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(-1, (short)301, 0, (short)0, 50103, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(3, (short)310, 0, (short)0, 50105, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(2, (short)335, 0, (short)0, 50105, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(3, (short)340, 0, (short)0, 50105, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(2, (short)365, 0, (short)0, 50105, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(3, (short)370, 0, (short)0, 50105, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(1, (short)391, 0, (short)0, 50105, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(-2, (short)395, 0, (short)0, 50105, Constants.EASTBOUND)); expectedResult.add(new StoppedCarTuple(-3, (short)400, 0, (short)0, 50105, Constants.EASTBOUND)); Assert.assertEquals(2, collector.output.size()); Assert.assertEquals(expectedFlushs, collector.output.get(TimestampMerger.FLUSH_STREAM_ID)); Assert.assertEquals(expectedResult, collector.output.get(Utils.DEFAULT_STREAM_ID)); Tuple flushTuple = mock(Tuple.class); when(flushTuple.getSourceStreamId()).thenReturn(TimestampMerger.FLUSH_STREAM_ID); bolt.execute(flushTuple); Assert.assertEquals(expectedFlushs.size() + 1, collector.output.get(TimestampMerger.FLUSH_STREAM_ID).size()); Assert.assertEquals(new Values((Object)null), collector.output.get(TimestampMerger.FLUSH_STREAM_ID).get(expectedFlushs.size())); }
@Override public void execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { if(input.getValue(0) == null) { 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()); Integer vid = this.inputPositionReport.getVid(); Short time = this.inputPositionReport.getTime(); int minute = Time.getMinute(time.longValue()); assert (minute >= this.currentMinute); if(minute > this.currentMinute) { LOGGER.trace("New minute: {}", new Integer(minute)); this.currentMinute = minute; this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values(time)); } if(this.inputPositionReport.isOnExitLane()) { this.lastPositions.remove(vid); this.collector.ack(input); return; } List<PositionReport> vehiclePositions = this.lastPositions.get(vid); if(vehiclePositions == null) { vehiclePositions = new LinkedList<PositionReport>(); vehiclePositions.add(this.inputPositionReport.copy()); this.lastPositions.put(vid, vehiclePositions); this.collector.ack(input); return; } this.lastPositionReport.clear(); this.lastPositionReport.addAll(vehiclePositions.get(0)); assert (this.inputPositionReport.getTime().shortValue() == this.lastPositionReport.getTime().shortValue() + 30); this.vehiclePosition.set(this.inputPositionReport); this.lastVehiclePosition.set(this.lastPositionReport); if(this.vehiclePosition.equals(this.lastVehiclePosition)) { vehiclePositions.add(0, this.inputPositionReport.copy()); if(vehiclePositions.size() >= STOPPED) { if(vehiclePositions.size() > STOPPED) { assert (vehiclePositions.size() == STOPPED + 1); vehiclePositions.remove(STOPPED); } this.collector.emit(new StoppedCarTuple(vid, time, this.inputPositionReport.getXWay(), this.inputPositionReport.getLane(), this.inputPositionReport.getPosition(), this.inputPositionReport.getDirection())); } this.collector.ack(input); return; } if(vehiclePositions.size() == STOPPED) { this.collector.emit(new StoppedCarTuple(new Integer(-vid.intValue()), time, this.inputPositionReport .getXWay(), this.lastPositionReport.getLane(), this.lastPositionReport.getPosition(), this.lastPositionReport.getDirection())); } vehiclePositions.clear(); vehiclePositions.add(this.inputPositionReport.copy()); this.collector.ack(input); }
StoppedCarsBolt extends BaseRichBolt { @Override public void execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { if(input.getValue(0) == null) { 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()); Integer vid = this.inputPositionReport.getVid(); Short time = this.inputPositionReport.getTime(); int minute = Time.getMinute(time.longValue()); assert (minute >= this.currentMinute); if(minute > this.currentMinute) { LOGGER.trace("New minute: {}", new Integer(minute)); this.currentMinute = minute; this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values(time)); } if(this.inputPositionReport.isOnExitLane()) { this.lastPositions.remove(vid); this.collector.ack(input); return; } List<PositionReport> vehiclePositions = this.lastPositions.get(vid); if(vehiclePositions == null) { vehiclePositions = new LinkedList<PositionReport>(); vehiclePositions.add(this.inputPositionReport.copy()); this.lastPositions.put(vid, vehiclePositions); this.collector.ack(input); return; } this.lastPositionReport.clear(); this.lastPositionReport.addAll(vehiclePositions.get(0)); assert (this.inputPositionReport.getTime().shortValue() == this.lastPositionReport.getTime().shortValue() + 30); this.vehiclePosition.set(this.inputPositionReport); this.lastVehiclePosition.set(this.lastPositionReport); if(this.vehiclePosition.equals(this.lastVehiclePosition)) { vehiclePositions.add(0, this.inputPositionReport.copy()); if(vehiclePositions.size() >= STOPPED) { if(vehiclePositions.size() > STOPPED) { assert (vehiclePositions.size() == STOPPED + 1); vehiclePositions.remove(STOPPED); } this.collector.emit(new StoppedCarTuple(vid, time, this.inputPositionReport.getXWay(), this.inputPositionReport.getLane(), this.inputPositionReport.getPosition(), this.inputPositionReport.getDirection())); } this.collector.ack(input); return; } if(vehiclePositions.size() == STOPPED) { this.collector.emit(new StoppedCarTuple(new Integer(-vid.intValue()), time, this.inputPositionReport .getXWay(), this.lastPositionReport.getLane(), this.lastPositionReport.getPosition(), this.lastPositionReport.getDirection())); } vehiclePositions.clear(); vehiclePositions.add(this.inputPositionReport.copy()); this.collector.ack(input); } }
StoppedCarsBolt extends BaseRichBolt { @Override public void execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { if(input.getValue(0) == null) { 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()); Integer vid = this.inputPositionReport.getVid(); Short time = this.inputPositionReport.getTime(); int minute = Time.getMinute(time.longValue()); assert (minute >= this.currentMinute); if(minute > this.currentMinute) { LOGGER.trace("New minute: {}", new Integer(minute)); this.currentMinute = minute; this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values(time)); } if(this.inputPositionReport.isOnExitLane()) { this.lastPositions.remove(vid); this.collector.ack(input); return; } List<PositionReport> vehiclePositions = this.lastPositions.get(vid); if(vehiclePositions == null) { vehiclePositions = new LinkedList<PositionReport>(); vehiclePositions.add(this.inputPositionReport.copy()); this.lastPositions.put(vid, vehiclePositions); this.collector.ack(input); return; } this.lastPositionReport.clear(); this.lastPositionReport.addAll(vehiclePositions.get(0)); assert (this.inputPositionReport.getTime().shortValue() == this.lastPositionReport.getTime().shortValue() + 30); this.vehiclePosition.set(this.inputPositionReport); this.lastVehiclePosition.set(this.lastPositionReport); if(this.vehiclePosition.equals(this.lastVehiclePosition)) { vehiclePositions.add(0, this.inputPositionReport.copy()); if(vehiclePositions.size() >= STOPPED) { if(vehiclePositions.size() > STOPPED) { assert (vehiclePositions.size() == STOPPED + 1); vehiclePositions.remove(STOPPED); } this.collector.emit(new StoppedCarTuple(vid, time, this.inputPositionReport.getXWay(), this.inputPositionReport.getLane(), this.inputPositionReport.getPosition(), this.inputPositionReport.getDirection())); } this.collector.ack(input); return; } if(vehiclePositions.size() == STOPPED) { this.collector.emit(new StoppedCarTuple(new Integer(-vid.intValue()), time, this.inputPositionReport .getXWay(), this.lastPositionReport.getLane(), this.lastPositionReport.getPosition(), this.lastPositionReport.getDirection())); } vehiclePositions.clear(); vehiclePositions.add(this.inputPositionReport.copy()); this.collector.ack(input); }  }
StoppedCarsBolt extends BaseRichBolt { @Override public void execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { if(input.getValue(0) == null) { 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()); Integer vid = this.inputPositionReport.getVid(); Short time = this.inputPositionReport.getTime(); int minute = Time.getMinute(time.longValue()); assert (minute >= this.currentMinute); if(minute > this.currentMinute) { LOGGER.trace("New minute: {}", new Integer(minute)); this.currentMinute = minute; this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values(time)); } if(this.inputPositionReport.isOnExitLane()) { this.lastPositions.remove(vid); this.collector.ack(input); return; } List<PositionReport> vehiclePositions = this.lastPositions.get(vid); if(vehiclePositions == null) { vehiclePositions = new LinkedList<PositionReport>(); vehiclePositions.add(this.inputPositionReport.copy()); this.lastPositions.put(vid, vehiclePositions); this.collector.ack(input); return; } this.lastPositionReport.clear(); this.lastPositionReport.addAll(vehiclePositions.get(0)); assert (this.inputPositionReport.getTime().shortValue() == this.lastPositionReport.getTime().shortValue() + 30); this.vehiclePosition.set(this.inputPositionReport); this.lastVehiclePosition.set(this.lastPositionReport); if(this.vehiclePosition.equals(this.lastVehiclePosition)) { vehiclePositions.add(0, this.inputPositionReport.copy()); if(vehiclePositions.size() >= STOPPED) { if(vehiclePositions.size() > STOPPED) { assert (vehiclePositions.size() == STOPPED + 1); vehiclePositions.remove(STOPPED); } this.collector.emit(new StoppedCarTuple(vid, time, this.inputPositionReport.getXWay(), this.inputPositionReport.getLane(), this.inputPositionReport.getPosition(), this.inputPositionReport.getDirection())); } this.collector.ack(input); return; } if(vehiclePositions.size() == STOPPED) { this.collector.emit(new StoppedCarTuple(new Integer(-vid.intValue()), time, this.inputPositionReport .getXWay(), this.lastPositionReport.getLane(), this.lastPositionReport.getPosition(), this.lastPositionReport.getDirection())); } vehiclePositions.clear(); vehiclePositions.add(this.inputPositionReport.copy()); this.collector.ack(input); }  @Override void prepare(@SuppressWarnings("rawtypes") Map conf, TopologyContext context, OutputCollector collector); @Override void execute(Tuple input); @Override void declareOutputFields(OutputFieldsDeclarer declarer); }
StoppedCarsBolt extends BaseRichBolt { @Override public void execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { if(input.getValue(0) == null) { 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()); Integer vid = this.inputPositionReport.getVid(); Short time = this.inputPositionReport.getTime(); int minute = Time.getMinute(time.longValue()); assert (minute >= this.currentMinute); if(minute > this.currentMinute) { LOGGER.trace("New minute: {}", new Integer(minute)); this.currentMinute = minute; this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values(time)); } if(this.inputPositionReport.isOnExitLane()) { this.lastPositions.remove(vid); this.collector.ack(input); return; } List<PositionReport> vehiclePositions = this.lastPositions.get(vid); if(vehiclePositions == null) { vehiclePositions = new LinkedList<PositionReport>(); vehiclePositions.add(this.inputPositionReport.copy()); this.lastPositions.put(vid, vehiclePositions); this.collector.ack(input); return; } this.lastPositionReport.clear(); this.lastPositionReport.addAll(vehiclePositions.get(0)); assert (this.inputPositionReport.getTime().shortValue() == this.lastPositionReport.getTime().shortValue() + 30); this.vehiclePosition.set(this.inputPositionReport); this.lastVehiclePosition.set(this.lastPositionReport); if(this.vehiclePosition.equals(this.lastVehiclePosition)) { vehiclePositions.add(0, this.inputPositionReport.copy()); if(vehiclePositions.size() >= STOPPED) { if(vehiclePositions.size() > STOPPED) { assert (vehiclePositions.size() == STOPPED + 1); vehiclePositions.remove(STOPPED); } this.collector.emit(new StoppedCarTuple(vid, time, this.inputPositionReport.getXWay(), this.inputPositionReport.getLane(), this.inputPositionReport.getPosition(), this.inputPositionReport.getDirection())); } this.collector.ack(input); return; } if(vehiclePositions.size() == STOPPED) { this.collector.emit(new StoppedCarTuple(new Integer(-vid.intValue()), time, this.inputPositionReport .getXWay(), this.lastPositionReport.getLane(), this.lastPositionReport.getPosition(), this.lastPositionReport.getDirection())); } vehiclePositions.clear(); vehiclePositions.add(this.inputPositionReport.copy()); this.collector.ack(input); }  @Override void prepare(@SuppressWarnings("rawtypes") Map conf, TopologyContext context, OutputCollector collector); @Override void execute(Tuple input); @Override void declareOutputFields(OutputFieldsDeclarer declarer);  }