@Test public void testExecute() { TestOutputCollector collector = new TestOutputCollector(); collector.output.put(Utils.DEFAULT_STREAM_ID, new LinkedList<List<Object>>()); AverageSpeedBolt bolt = new AverageSpeedBolt(); bolt.prepare(null, null, new OutputCollector(collector)); final Tuple tuple = mock(Tuple.class); when(tuple.getSourceStreamId()).thenReturn("streamId"); OngoingStubbing<List<Object>> tupleStub = when(tuple.getValues()); final int numberOfVehicles = 1 + this.r.nextInt(50); Integer[] vids = new Integer[numberOfVehicles]; for(int i = 0; i < numberOfVehicles; ++i) { vids[i] = new Integer(this.r.nextInt(Integer.MAX_VALUE)); for(int j = 0; j < i; ++j) { if(vids[i].intValue() == vids[j].intValue()) { --i; break; } } } Short[] segment = new Short[numberOfVehicles]; for(int i = 0; i < numberOfVehicles; ++i) { segment[i] = new Short((short)(Constants.MAX_SEGMENT / 3 + this.r.nextInt(Constants.MAX_SEGMENT / 3))); } final int numberOfMinutes = 2 + this.r.nextInt(8); @SuppressWarnings("unchecked") final HashMap<Short, List<Integer>>[] speedValuesPerSegment1 = new HashMap[numberOfMinutes]; @SuppressWarnings("unchecked") final HashMap<Short, List<Integer>>[] speedValuesPerSegment2 = new HashMap[numberOfMinutes]; for(short minute = 1; minute <= numberOfMinutes; ++minute) { speedValuesPerSegment1[minute - 1] = new HashMap<Short, List<Integer>>(); speedValuesPerSegment2[minute - 1] = new HashMap<Short, List<Integer>>(); LinkedList<Integer> index = new LinkedList<Integer>(); for(int i = 0; i < numberOfVehicles; ++i) { index.add(new Integer(i)); } while(index.size() > 0) { int idx = index.remove(this.r.nextInt(index.size())).intValue(); int speed = this.r.nextInt(Constants.MAX_SPEED); if(idx < numberOfVehicles / 2) { tupleStub = tupleStub.thenReturn(new AvgVehicleSpeedTuple(vids[idx], new Short(minute), new Integer(1), segment[idx], new Short((short)0), new Double(speed))); this.addToSpeedList(speedValuesPerSegment1[minute - 1], segment[idx], speed); if(this.r.nextInt(5) > 0) { segment[idx] = new Short((short)(segment[idx].shortValue() + 1)); } } else { tupleStub = tupleStub.thenReturn(new AvgVehicleSpeedTuple(vids[idx], new Short(minute), new Integer(1), segment[idx], new Short((short)1), new Double(speed))); this.addToSpeedList(speedValuesPerSegment2[minute - 1], segment[idx], speed); if(this.r.nextInt(5) > 0) { segment[idx] = new Short((short)(segment[idx].shortValue() - 1)); } } } } final HashSet<AvgSpeedTuple> expectedResult = new HashSet<AvgSpeedTuple>(); final LinkedList<Values> expectedFlushs = new LinkedList<Values>(); for(short minute = 1; minute <= numberOfMinutes; ++minute) { for(int i = 0; i < numberOfVehicles; ++i) { bolt.execute(tuple); } expectedFlushs.add(new Values(new Short(minute))); Assert.assertEquals(2, collector.output.size()); Assert.assertEquals(expectedResult, Sets.newHashSet(collector.output.get(Utils.DEFAULT_STREAM_ID))); Assert.assertEquals(expectedFlushs, collector.output.get(TimestampMerger.FLUSH_STREAM_ID)); Assert.assertEquals(minute * numberOfVehicles, collector.acked.size()); Assert.assertEquals(0, collector.failed.size()); collector.output.clear(); collector.output.put(Utils.DEFAULT_STREAM_ID, new LinkedList<List<Object>>()); expectedResult.clear(); expectedFlushs.clear(); this.addToExpectedResult(speedValuesPerSegment1[minute - 1], minute, (short)0, expectedResult); this.addToExpectedResult(speedValuesPerSegment2[minute - 1], minute, (short)1, expectedResult); } 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(2, collector.output.size()); Assert.assertEquals(expectedResult, Sets.newHashSet(collector.output.get(Utils.DEFAULT_STREAM_ID))); Assert.assertEquals(numberOfMinutes * numberOfVehicles + 1, collector.acked.size()); 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)); Assert.assertEquals(0, collector.failed.size()); }
@Override public void execute(Tuple input) { if(input.getSourceStreamId().equals(TimestampMerger.FLUSH_STREAM_ID)) { Object ts = input.getValue(0); if(ts == null) { this.flushBuffer(); this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values((Object)null)); } else { this.checkMinute(((Number)ts).shortValue()); } this.collector.ack(input); return; } this.inputTuple.clear(); this.inputTuple.addAll(input.getValues()); LOGGER.trace(this.inputTuple.toString()); this.checkMinute(this.inputTuple.getMinute().shortValue()); double avgVehicleSpeed = this.inputTuple.getAvgSpeed().doubleValue(); this.segment.set(this.inputTuple); AvgValue segAvg = this.avgSpeedsMap.get(this.segment); if(segAvg == null) { segAvg = new AvgValue(avgVehicleSpeed); this.avgSpeedsMap.put(this.segment.copy(), segAvg); } else { segAvg.updateAverage(avgVehicleSpeed); } this.collector.ack(input); }
AverageSpeedBolt 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.flushBuffer(); this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values((Object)null)); } else { this.checkMinute(((Number)ts).shortValue()); } this.collector.ack(input); return; } this.inputTuple.clear(); this.inputTuple.addAll(input.getValues()); LOGGER.trace(this.inputTuple.toString()); this.checkMinute(this.inputTuple.getMinute().shortValue()); double avgVehicleSpeed = this.inputTuple.getAvgSpeed().doubleValue(); this.segment.set(this.inputTuple); AvgValue segAvg = this.avgSpeedsMap.get(this.segment); if(segAvg == null) { segAvg = new AvgValue(avgVehicleSpeed); this.avgSpeedsMap.put(this.segment.copy(), segAvg); } else { segAvg.updateAverage(avgVehicleSpeed); } this.collector.ack(input); } }
AverageSpeedBolt 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.flushBuffer(); this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values((Object)null)); } else { this.checkMinute(((Number)ts).shortValue()); } this.collector.ack(input); return; } this.inputTuple.clear(); this.inputTuple.addAll(input.getValues()); LOGGER.trace(this.inputTuple.toString()); this.checkMinute(this.inputTuple.getMinute().shortValue()); double avgVehicleSpeed = this.inputTuple.getAvgSpeed().doubleValue(); this.segment.set(this.inputTuple); AvgValue segAvg = this.avgSpeedsMap.get(this.segment); if(segAvg == null) { segAvg = new AvgValue(avgVehicleSpeed); this.avgSpeedsMap.put(this.segment.copy(), segAvg); } else { segAvg.updateAverage(avgVehicleSpeed); } this.collector.ack(input); }  }
AverageSpeedBolt 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.flushBuffer(); this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values((Object)null)); } else { this.checkMinute(((Number)ts).shortValue()); } this.collector.ack(input); return; } this.inputTuple.clear(); this.inputTuple.addAll(input.getValues()); LOGGER.trace(this.inputTuple.toString()); this.checkMinute(this.inputTuple.getMinute().shortValue()); double avgVehicleSpeed = this.inputTuple.getAvgSpeed().doubleValue(); this.segment.set(this.inputTuple); AvgValue segAvg = this.avgSpeedsMap.get(this.segment); if(segAvg == null) { segAvg = new AvgValue(avgVehicleSpeed); this.avgSpeedsMap.put(this.segment.copy(), segAvg); } else { segAvg.updateAverage(avgVehicleSpeed); } 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); }
AverageSpeedBolt 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.flushBuffer(); this.collector.emit(TimestampMerger.FLUSH_STREAM_ID, new Values((Object)null)); } else { this.checkMinute(((Number)ts).shortValue()); } this.collector.ack(input); return; } this.inputTuple.clear(); this.inputTuple.addAll(input.getValues()); LOGGER.trace(this.inputTuple.toString()); this.checkMinute(this.inputTuple.getMinute().shortValue()); double avgVehicleSpeed = this.inputTuple.getAvgSpeed().doubleValue(); this.segment.set(this.inputTuple); AvgValue segAvg = this.avgSpeedsMap.get(this.segment); if(segAvg == null) { segAvg = new AvgValue(avgVehicleSpeed); this.avgSpeedsMap.put(this.segment.copy(), segAvg); } else { segAvg.updateAverage(avgVehicleSpeed); } 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);  }