@Test public void testExecute() { AverageVehicleSpeedBolt bolt = new AverageVehicleSpeedBolt(); TestOutputCollector collector = new TestOutputCollector(); bolt.prepare(null, null, new OutputCollector(collector)); final Tuple tuple = mock(Tuple.class); int vehicleID0, vehicleID1, vehicleID2; vehicleID0 = (int)(this.r.nextDouble() * Constants.NUMBER_OF_VIDS); do { vehicleID1 = (int)(this.r.nextDouble() * Constants.NUMBER_OF_VIDS); } while(vehicleID1 == vehicleID0); do { vehicleID2 = (int)(this.r.nextDouble() * Constants.NUMBER_OF_VIDS); } while(vehicleID2 == vehicleID1 || vehicleID2 == vehicleID0); short segment = (short)(this.r.nextDouble() * Constants.NUMBER_OF_SEGMENT); short time0 = (short)this.r.nextInt(60); PositionReport posReport0Stopped = EntityHelper.createPosReport(new Short(time0), new Short(segment), this.r, new Integer(vehicleID0), 0, 0 ); when(tuple.getValues()).thenReturn(posReport0Stopped); when(tuple.getSourceStreamId()).thenReturn("streamId"); bolt.execute(tuple); assertEquals(1, collector.acked.size()); assertEquals(0, collector.output.size()); short time1 = (short)(time0 + 60); PositionReport posReport1Stopped = EntityHelper.createPosReport(new Short(time1), new Short(segment), this.r, new Integer(vehicleID0), 0, 0 ); when(tuple.getValues()).thenReturn(posReport1Stopped); when(tuple.getSourceStreamId()).thenReturn("streamId"); bolt.execute(tuple); assertEquals(2, collector.acked.size()); assertEquals(1, collector.output.get(Utils.DEFAULT_STREAM_ID).size()); AvgVehicleSpeedTuple result = (AvgVehicleSpeedTuple)collector.output.get(Utils.DEFAULT_STREAM_ID).get(0); assertEquals(new SegmentIdentifier(posReport1Stopped), new SegmentIdentifier(result)); assertEquals(time1 / 60, result.getMinute().shortValue()); assertEquals(0.0, result.getAvgSpeed().doubleValue(), 0.0); assertEquals(1, collector.output.get(TimestampMerger.FLUSH_STREAM_ID).size()); assertEquals(new Values(new Short(Time.getMinute(time1))), collector.output .get(TimestampMerger.FLUSH_STREAM_ID).get(0)); int speed2 = this.r.nextInt(Constants.NUMBER_OF_SPEEDS); int speed3 = this.r.nextInt(Constants.NUMBER_OF_SPEEDS); int speed4 = this.r.nextInt(Constants.NUMBER_OF_SPEEDS); PositionReport posReport2Running = EntityHelper.createPosReport(new Short(time1), new Short(segment), this.r, new Integer(vehicleID1), speed2, speed2 ); when(tuple.getValues()).thenReturn(posReport2Running); when(tuple.getSourceStreamId()).thenReturn("streamId"); bolt.execute(tuple); assertEquals(3, collector.acked.size()); assertEquals(1, collector.output.get(Utils.DEFAULT_STREAM_ID).size()); assertEquals(1, collector.output.get(TimestampMerger.FLUSH_STREAM_ID).size()); PositionReport posReport3Running = EntityHelper.createPosReport(new Short(time1), new Short(segment), this.r, new Integer(vehicleID1), speed3, speed3 ); when(tuple.getValues()).thenReturn(posReport3Running); when(tuple.getSourceStreamId()).thenReturn("streamId"); bolt.execute(tuple); assertEquals(4, collector.acked.size()); assertEquals(1, collector.output.get(Utils.DEFAULT_STREAM_ID).size()); assertEquals(1, collector.output.get(TimestampMerger.FLUSH_STREAM_ID).size()); short time2 = (short)(time1 + 60); PositionReport posReport4Running = EntityHelper.createPosReport(new Short(time2), new Short(segment), this.r, new Integer(vehicleID2), speed4, speed4 ); when(tuple.getValues()).thenReturn(posReport4Running); when(tuple.getSourceStreamId()).thenReturn("streamId"); bolt.execute(tuple); assertEquals(5, collector.acked.size()); assertEquals(3, collector.output.get(Utils.DEFAULT_STREAM_ID).size()); assertEquals(2, collector.output.get(TimestampMerger.FLUSH_STREAM_ID).size()); assertEquals(new Values(new Short(Time.getMinute(time2))), collector.output .get(TimestampMerger.FLUSH_STREAM_ID).get(1)); for(int i = 1; i < 3; ++i) { result = (AvgVehicleSpeedTuple)collector.output.get(Utils.DEFAULT_STREAM_ID).get(i); assertEquals(new SegmentIdentifier(posReport1Stopped), new SegmentIdentifier(result)); if(result.getVid().intValue() == vehicleID0) { assertEquals(0.0, result.getAvgSpeed().doubleValue(), 0.0); } else { assertEquals(result.getVid().intValue(), vehicleID1); assertEquals((speed2 + speed3) / 2.0, result.getAvgSpeed().doubleValue(), 0.0); } assertEquals(time2 / 60, result.getMinute().shortValue()); } Tuple flushTuple = mock(Tuple.class); when(flushTuple.getSourceStreamId()).thenReturn(TimestampMerger.FLUSH_STREAM_ID); bolt.execute(flushTuple); result = (AvgVehicleSpeedTuple)collector.output.get(Utils.DEFAULT_STREAM_ID).get(3); assertEquals(new SegmentIdentifier(posReport1Stopped), new SegmentIdentifier(result)); assertEquals(result.getVid().intValue(), vehicleID2); assertEquals(speed4, result.getAvgSpeed().doubleValue(), 0.0); assertEquals((time2 + 60) / 60, result.getMinute().shortValue()); assertEquals(3, collector.output.get(TimestampMerger.FLUSH_STREAM_ID).size()); assertEquals(new Values((Object)null), collector.output.get(TimestampMerger.FLUSH_STREAM_ID).get(2)); assertEquals(2, collector.output.size()); }
@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()); Integer vid = this.inputPositionReport.getVid(); short minute = this.inputPositionReport.getMinuteNumber(); int speed = this.inputPositionReport.getSpeed().intValue(); this.segment.set(this.inputPositionReport); assert (minute >= this.currentMinute); if(minute > this.currentMinute) { this.flushBuffer(); this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values(new Short(minute))); this.avgSpeedsMap.clear(); this.currentMinute = minute; } Pair<AvgValue, SegmentIdentifier> vehicleEntry = this.avgSpeedsMap.get(vid); if(vehicleEntry != null && !vehicleEntry.getRight().equals(this.segment)) { SegmentIdentifier segId = vehicleEntry.getRight(); this.collector.emit(new AvgVehicleSpeedTuple(vid, new Short(this.currentMinute), segId.getXWay(), segId .getSegment(), segId.getDirection(), vehicleEntry.getLeft().getAverage())); vehicleEntry = null; } if(vehicleEntry == null) { vehicleEntry = new MutablePair<AvgValue, SegmentIdentifier>(new AvgValue(speed), this.segment.copy()); this.avgSpeedsMap.put(vid, vehicleEntry); } else { vehicleEntry.getLeft().updateAverage(speed); } this.collector.ack(input); }
AverageVehicleSpeedBolt 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()); Integer vid = this.inputPositionReport.getVid(); short minute = this.inputPositionReport.getMinuteNumber(); int speed = this.inputPositionReport.getSpeed().intValue(); this.segment.set(this.inputPositionReport); assert (minute >= this.currentMinute); if(minute > this.currentMinute) { this.flushBuffer(); this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values(new Short(minute))); this.avgSpeedsMap.clear(); this.currentMinute = minute; } Pair<AvgValue, SegmentIdentifier> vehicleEntry = this.avgSpeedsMap.get(vid); if(vehicleEntry != null && !vehicleEntry.getRight().equals(this.segment)) { SegmentIdentifier segId = vehicleEntry.getRight(); this.collector.emit(new AvgVehicleSpeedTuple(vid, new Short(this.currentMinute), segId.getXWay(), segId .getSegment(), segId.getDirection(), vehicleEntry.getLeft().getAverage())); vehicleEntry = null; } if(vehicleEntry == null) { vehicleEntry = new MutablePair<AvgValue, SegmentIdentifier>(new AvgValue(speed), this.segment.copy()); this.avgSpeedsMap.put(vid, vehicleEntry); } else { vehicleEntry.getLeft().updateAverage(speed); } this.collector.ack(input); } }
AverageVehicleSpeedBolt 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()); Integer vid = this.inputPositionReport.getVid(); short minute = this.inputPositionReport.getMinuteNumber(); int speed = this.inputPositionReport.getSpeed().intValue(); this.segment.set(this.inputPositionReport); assert (minute >= this.currentMinute); if(minute > this.currentMinute) { this.flushBuffer(); this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values(new Short(minute))); this.avgSpeedsMap.clear(); this.currentMinute = minute; } Pair<AvgValue, SegmentIdentifier> vehicleEntry = this.avgSpeedsMap.get(vid); if(vehicleEntry != null && !vehicleEntry.getRight().equals(this.segment)) { SegmentIdentifier segId = vehicleEntry.getRight(); this.collector.emit(new AvgVehicleSpeedTuple(vid, new Short(this.currentMinute), segId.getXWay(), segId .getSegment(), segId.getDirection(), vehicleEntry.getLeft().getAverage())); vehicleEntry = null; } if(vehicleEntry == null) { vehicleEntry = new MutablePair<AvgValue, SegmentIdentifier>(new AvgValue(speed), this.segment.copy()); this.avgSpeedsMap.put(vid, vehicleEntry); } else { vehicleEntry.getLeft().updateAverage(speed); } this.collector.ack(input); }  }
AverageVehicleSpeedBolt 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()); Integer vid = this.inputPositionReport.getVid(); short minute = this.inputPositionReport.getMinuteNumber(); int speed = this.inputPositionReport.getSpeed().intValue(); this.segment.set(this.inputPositionReport); assert (minute >= this.currentMinute); if(minute > this.currentMinute) { this.flushBuffer(); this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values(new Short(minute))); this.avgSpeedsMap.clear(); this.currentMinute = minute; } Pair<AvgValue, SegmentIdentifier> vehicleEntry = this.avgSpeedsMap.get(vid); if(vehicleEntry != null && !vehicleEntry.getRight().equals(this.segment)) { SegmentIdentifier segId = vehicleEntry.getRight(); this.collector.emit(new AvgVehicleSpeedTuple(vid, new Short(this.currentMinute), segId.getXWay(), segId .getSegment(), segId.getDirection(), vehicleEntry.getLeft().getAverage())); vehicleEntry = null; } if(vehicleEntry == null) { vehicleEntry = new MutablePair<AvgValue, SegmentIdentifier>(new AvgValue(speed), this.segment.copy()); this.avgSpeedsMap.put(vid, vehicleEntry); } else { vehicleEntry.getLeft().updateAverage(speed); } 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); }
AverageVehicleSpeedBolt 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()); Integer vid = this.inputPositionReport.getVid(); short minute = this.inputPositionReport.getMinuteNumber(); int speed = this.inputPositionReport.getSpeed().intValue(); this.segment.set(this.inputPositionReport); assert (minute >= this.currentMinute); if(minute > this.currentMinute) { this.flushBuffer(); this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values(new Short(minute))); this.avgSpeedsMap.clear(); this.currentMinute = minute; } Pair<AvgValue, SegmentIdentifier> vehicleEntry = this.avgSpeedsMap.get(vid); if(vehicleEntry != null && !vehicleEntry.getRight().equals(this.segment)) { SegmentIdentifier segId = vehicleEntry.getRight(); this.collector.emit(new AvgVehicleSpeedTuple(vid, new Short(this.currentMinute), segId.getXWay(), segId .getSegment(), segId.getDirection(), vehicleEntry.getLeft().getAverage())); vehicleEntry = null; } if(vehicleEntry == null) { vehicleEntry = new MutablePair<AvgValue, SegmentIdentifier>(new AvgValue(speed), this.segment.copy()); this.avgSpeedsMap.put(vid, vehicleEntry); } else { vehicleEntry.getLeft().updateAverage(speed); } 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);  }