@SuppressWarnings("boxing") @Test public void testExecute() { Tuple tuple = mock(Tuple.class); when(tuple.getSourceStreamId()).thenReturn("streamId"); OngoingStubbing<List<Object>> tupleValueStub = when(tuple.getValues()); int[] m = new int[timeAndPositions.length]; int tupleCount = 0; while(true) { int vid = -1; int minTs = Integer.MAX_VALUE; for(int i = 0; i < timeAndPositions.length; ++i) { if(m[i] < timeAndPositions[i].length) { if(timeAndPositions[i][m[i]][1] < minTs) { minTs = timeAndPositions[i][m[i]][1]; vid = i; } } } if(minTs == Integer.MAX_VALUE) { break; } StoppedCarTuple t = new StoppedCarTuple((vid + 1) * timeAndPositions[vid][m[vid]][3], timeAndPositions[vid][m[vid]][1].shortValue(), 0, timeAndPositions[vid][m[vid]][0].shortValue(), timeAndPositions[vid][m[vid]][2], Constants.EASTBOUND); tupleValueStub = tupleValueStub.thenReturn(t); ++tupleCount; ++m[vid]; } tupleValueStub.thenReturn(null); AccidentDetectionBolt bolt = new AccidentDetectionBolt(); TestOutputCollector collector = new TestOutputCollector(); bolt.prepare(null, null, new OutputCollector(collector)); for(int i = 0; i < tupleCount; ++i) { bolt.execute(tuple); } Tuple flushTuple = mock(Tuple.class); when(flushTuple.getSourceStreamId()).thenReturn(TimestampMerger.FLUSH_STREAM_ID); bolt.execute(flushTuple); List<Values> expectedFlushes = new ArrayList<Values>(); expectedFlushes.add(new Values((short)119)); expectedFlushes.add(new Values((short)179)); expectedFlushes.add(new Values((short)239)); expectedFlushes.add(new Values((short)299)); expectedFlushes.add(new Values((short)359)); expectedFlushes.add(new Values((short)419)); expectedFlushes.add(new Values((short)479)); expectedFlushes.add(new Values((Object)null)); List<Values> expectedResult = new ArrayList<Values>(); expectedResult.add(new AccidentTuple((short)425, 0, (short)9, (short)0)); expectedResult.add(new AccidentTuple((short)430, 0, (short)9, (short)0)); expectedResult.add(new AccidentTuple((short)455, 0, (short)9, (short)0)); expectedResult.add(new AccidentTuple((short)460, 0, (short)9, (short)0)); expectedResult.add(new AccidentTuple((short)481, 0, (short)9, (short)0)); Assert.assertEquals(2, collector.output.size()); Assert.assertEquals(expectedResult, collector.output.get(TopologyControl.ACCIDENTS_STREAM_ID)); Assert.assertEquals(expectedFlushes, collector.output.get(TimestampMerger.FLUSH_STREAM_ID)); }
@Override public void execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { Object ts = input.getValue(0); if(ts == null) { this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values((Object)null)); } else { this.checkMinute(Time.getMinute(((Number)ts).longValue())); } this.collector.ack(input); return; } this.inputStoppedCarTuple.clear(); this.inputStoppedCarTuple.addAll(input.getValues()); LOGGER.trace(this.inputStoppedCarTuple.toString()); this.checkMinute(this.inputStoppedCarTuple.getMinuteNumber()); assert (this.inputStoppedCarTuple.getLane().shortValue() != Constants.EXIT_LANE); Integer vid = this.inputStoppedCarTuple.getVid(); int v = vid.intValue(); boolean isStopReport = v > 0; this.stoppedVehiclePosition.set(this.inputStoppedCarTuple); Set<Integer> stoppedCars = this.stoppedCarsPerPosition.get(this.stoppedVehiclePosition); if(isStopReport) { if(stoppedCars == null) { stoppedCars = new HashSet<Integer>(); stoppedCars.add(vid); this.stoppedCarsPerPosition.put(this.stoppedVehiclePosition.copy(), stoppedCars); } else { stoppedCars.add(vid); if(stoppedCars.size() > 1) { this.collector.emit(TopologyControl.ACCIDENTS_STREAM_ID, new AccidentTuple( this.inputStoppedCarTuple.getTime(), this.inputStoppedCarTuple.getXWay(), this.inputStoppedCarTuple.getSegment(), this.inputStoppedCarTuple.getDirection())); } } } else { stoppedCars.remove(new Integer(-v)); if(stoppedCars.isEmpty()) { this.stoppedCarsPerPosition.remove(this.inputStoppedCarTuple); } } this.collector.ack(input); }
AccidentDetectionBolt extends BaseRichBolt { @Override public void execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { Object ts = input.getValue(0); if(ts == null) { this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values((Object)null)); } else { this.checkMinute(Time.getMinute(((Number)ts).longValue())); } this.collector.ack(input); return; } this.inputStoppedCarTuple.clear(); this.inputStoppedCarTuple.addAll(input.getValues()); LOGGER.trace(this.inputStoppedCarTuple.toString()); this.checkMinute(this.inputStoppedCarTuple.getMinuteNumber()); assert (this.inputStoppedCarTuple.getLane().shortValue() != Constants.EXIT_LANE); Integer vid = this.inputStoppedCarTuple.getVid(); int v = vid.intValue(); boolean isStopReport = v > 0; this.stoppedVehiclePosition.set(this.inputStoppedCarTuple); Set<Integer> stoppedCars = this.stoppedCarsPerPosition.get(this.stoppedVehiclePosition); if(isStopReport) { if(stoppedCars == null) { stoppedCars = new HashSet<Integer>(); stoppedCars.add(vid); this.stoppedCarsPerPosition.put(this.stoppedVehiclePosition.copy(), stoppedCars); } else { stoppedCars.add(vid); if(stoppedCars.size() > 1) { this.collector.emit(TopologyControl.ACCIDENTS_STREAM_ID, new AccidentTuple( this.inputStoppedCarTuple.getTime(), this.inputStoppedCarTuple.getXWay(), this.inputStoppedCarTuple.getSegment(), this.inputStoppedCarTuple.getDirection())); } } } else { stoppedCars.remove(new Integer(-v)); if(stoppedCars.isEmpty()) { this.stoppedCarsPerPosition.remove(this.inputStoppedCarTuple); } } this.collector.ack(input); } }
AccidentDetectionBolt extends BaseRichBolt { @Override public void execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { Object ts = input.getValue(0); if(ts == null) { this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values((Object)null)); } else { this.checkMinute(Time.getMinute(((Number)ts).longValue())); } this.collector.ack(input); return; } this.inputStoppedCarTuple.clear(); this.inputStoppedCarTuple.addAll(input.getValues()); LOGGER.trace(this.inputStoppedCarTuple.toString()); this.checkMinute(this.inputStoppedCarTuple.getMinuteNumber()); assert (this.inputStoppedCarTuple.getLane().shortValue() != Constants.EXIT_LANE); Integer vid = this.inputStoppedCarTuple.getVid(); int v = vid.intValue(); boolean isStopReport = v > 0; this.stoppedVehiclePosition.set(this.inputStoppedCarTuple); Set<Integer> stoppedCars = this.stoppedCarsPerPosition.get(this.stoppedVehiclePosition); if(isStopReport) { if(stoppedCars == null) { stoppedCars = new HashSet<Integer>(); stoppedCars.add(vid); this.stoppedCarsPerPosition.put(this.stoppedVehiclePosition.copy(), stoppedCars); } else { stoppedCars.add(vid); if(stoppedCars.size() > 1) { this.collector.emit(TopologyControl.ACCIDENTS_STREAM_ID, new AccidentTuple( this.inputStoppedCarTuple.getTime(), this.inputStoppedCarTuple.getXWay(), this.inputStoppedCarTuple.getSegment(), this.inputStoppedCarTuple.getDirection())); } } } else { stoppedCars.remove(new Integer(-v)); if(stoppedCars.isEmpty()) { this.stoppedCarsPerPosition.remove(this.inputStoppedCarTuple); } } this.collector.ack(input); }  }
AccidentDetectionBolt extends BaseRichBolt { @Override public void execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { Object ts = input.getValue(0); if(ts == null) { this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values((Object)null)); } else { this.checkMinute(Time.getMinute(((Number)ts).longValue())); } this.collector.ack(input); return; } this.inputStoppedCarTuple.clear(); this.inputStoppedCarTuple.addAll(input.getValues()); LOGGER.trace(this.inputStoppedCarTuple.toString()); this.checkMinute(this.inputStoppedCarTuple.getMinuteNumber()); assert (this.inputStoppedCarTuple.getLane().shortValue() != Constants.EXIT_LANE); Integer vid = this.inputStoppedCarTuple.getVid(); int v = vid.intValue(); boolean isStopReport = v > 0; this.stoppedVehiclePosition.set(this.inputStoppedCarTuple); Set<Integer> stoppedCars = this.stoppedCarsPerPosition.get(this.stoppedVehiclePosition); if(isStopReport) { if(stoppedCars == null) { stoppedCars = new HashSet<Integer>(); stoppedCars.add(vid); this.stoppedCarsPerPosition.put(this.stoppedVehiclePosition.copy(), stoppedCars); } else { stoppedCars.add(vid); if(stoppedCars.size() > 1) { this.collector.emit(TopologyControl.ACCIDENTS_STREAM_ID, new AccidentTuple( this.inputStoppedCarTuple.getTime(), this.inputStoppedCarTuple.getXWay(), this.inputStoppedCarTuple.getSegment(), this.inputStoppedCarTuple.getDirection())); } } } else { stoppedCars.remove(new Integer(-v)); if(stoppedCars.isEmpty()) { this.stoppedCarsPerPosition.remove(this.inputStoppedCarTuple); } } 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); }
AccidentDetectionBolt extends BaseRichBolt { @Override public void execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { Object ts = input.getValue(0); if(ts == null) { this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values((Object)null)); } else { this.checkMinute(Time.getMinute(((Number)ts).longValue())); } this.collector.ack(input); return; } this.inputStoppedCarTuple.clear(); this.inputStoppedCarTuple.addAll(input.getValues()); LOGGER.trace(this.inputStoppedCarTuple.toString()); this.checkMinute(this.inputStoppedCarTuple.getMinuteNumber()); assert (this.inputStoppedCarTuple.getLane().shortValue() != Constants.EXIT_LANE); Integer vid = this.inputStoppedCarTuple.getVid(); int v = vid.intValue(); boolean isStopReport = v > 0; this.stoppedVehiclePosition.set(this.inputStoppedCarTuple); Set<Integer> stoppedCars = this.stoppedCarsPerPosition.get(this.stoppedVehiclePosition); if(isStopReport) { if(stoppedCars == null) { stoppedCars = new HashSet<Integer>(); stoppedCars.add(vid); this.stoppedCarsPerPosition.put(this.stoppedVehiclePosition.copy(), stoppedCars); } else { stoppedCars.add(vid); if(stoppedCars.size() > 1) { this.collector.emit(TopologyControl.ACCIDENTS_STREAM_ID, new AccidentTuple( this.inputStoppedCarTuple.getTime(), this.inputStoppedCarTuple.getXWay(), this.inputStoppedCarTuple.getSegment(), this.inputStoppedCarTuple.getDirection())); } } } else { stoppedCars.remove(new Integer(-v)); if(stoppedCars.isEmpty()) { this.stoppedCarsPerPosition.remove(this.inputStoppedCarTuple); } } 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);  }