@SuppressWarnings("boxing") @Test public void testExecute() { Tuple tuple = mock(Tuple.class); OngoingStubbing<List<Object>> valueStub = when(tuple.getValues()); for(Short[] input : PosRepAndAccs) { if(input.length == 3) { valueStub = valueStub.thenReturn(new PositionReport(input[0], dummyInt, dummyInt, dummyInt, input[1], dummyShort, input[2], dummyInt)); } else { assert (input.length == 2); valueStub = valueStub.thenReturn(new AccidentTuple((short)((input[0] * 60) - 1), dummyInt, input[1], dummyShort)); } } valueStub.thenReturn(null); OngoingStubbing<String> streamStub = when(tuple.getSourceStreamId()); for(Short[] input : PosRepAndAccs) { if(input.length == 3) { streamStub = streamStub.thenReturn(TopologyControl.POSITION_REPORTS_STREAM_ID); } else { assert (input.length == 2); streamStub = streamStub.thenReturn(Utils.DEFAULT_STREAM_ID); } } streamStub.thenReturn(null); AccidentNotificationBolt bolt = new AccidentNotificationBolt(); TestOutputCollector collector = new TestOutputCollector(); bolt.prepare(null, null, new OutputCollector(collector)); for(int i = 0; i < PosRepAndAccs.length; ++i) { bolt.execute(tuple); } 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); List<AccidentNotification> expectedResult = new ArrayList<AccidentNotification>(); expectedResult.add(new AccidentNotification((short)121, (short)121, (short)2, dummyInt)); Assert.assertEquals(2, collector.output.size()); Assert.assertEquals(expectedResult, collector.output.get(Utils.DEFAULT_STREAM_ID)); 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)); }
@Override public void execute(Tuple input) { final String sourceStreamId = input.getSourceStreamId(); if(sourceStreamId.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).shortValue())); } this.collector.ack(input); return; } if(sourceStreamId.equals(TopologyControl.POSITION_REPORTS_STREAM_ID)) { this.inputPositionReport.clear(); this.inputPositionReport.addAll(input.getValues()); LOGGER.trace(this.inputPositionReport.toString()); this.checkMinute(this.inputPositionReport.getMinuteNumber()); if(this.inputPositionReport.isOnExitLane()) { this.collector.ack(input); return; } final Short currentSegment = this.inputPositionReport.getSegment(); final Integer vid = this.inputPositionReport.getVid(); final Short previousSegment = this.allCars.put(vid, currentSegment); if(previousSegment != null && currentSegment.shortValue() == previousSegment.shortValue()) { this.collector.ack(input); return; } final Short direction = this.inputPositionReport.getDirection(); final short dir = direction.shortValue(); final short diff = (short)-(dir - 1 + ((dir + 1) / 2)); assert (dir == Constants.EASTBOUND.shortValue() ? diff == 1 : diff == -1); final Integer xway = this.inputPositionReport.getXWay(); final short curSeg = currentSegment.shortValue(); this.segmentToCheck.setXWay(xway); this.segmentToCheck.setDirection(direction); for(int i = 0; i <= 4; ++i) { final short nextSegment = (short)(curSeg + (diff * i)); assert (dir == Constants.EASTBOUND.shortValue() ? nextSegment >= curSeg : nextSegment <= curSeg); this.segmentToCheck.setSegment(new Short(nextSegment)); if(this.previousMinuteAccidents.contains(this.segmentToCheck)) { this.collector.emit(new AccidentNotification(this.inputPositionReport.getTime(), this.inputPositionReport.getTime(), this.segmentToCheck.getSegment(), vid)); break; } } } else { this.inputAccidentTuple.clear(); this.inputAccidentTuple.addAll(input.getValues()); LOGGER.trace(this.inputAccidentTuple.toString()); this.checkMinute(this.inputAccidentTuple.getMinuteNumber()); assert (this.inputAccidentTuple.getMinuteNumber() == this.currentMinute); this.currentMinuteAccidents.add(new SegmentIdentifier(this.inputAccidentTuple)); } this.collector.ack(input); }
AccidentNotificationBolt extends BaseRichBolt { @Override public void execute(Tuple input) { final String sourceStreamId = input.getSourceStreamId(); if(sourceStreamId.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).shortValue())); } this.collector.ack(input); return; } if(sourceStreamId.equals(TopologyControl.POSITION_REPORTS_STREAM_ID)) { this.inputPositionReport.clear(); this.inputPositionReport.addAll(input.getValues()); LOGGER.trace(this.inputPositionReport.toString()); this.checkMinute(this.inputPositionReport.getMinuteNumber()); if(this.inputPositionReport.isOnExitLane()) { this.collector.ack(input); return; } final Short currentSegment = this.inputPositionReport.getSegment(); final Integer vid = this.inputPositionReport.getVid(); final Short previousSegment = this.allCars.put(vid, currentSegment); if(previousSegment != null && currentSegment.shortValue() == previousSegment.shortValue()) { this.collector.ack(input); return; } final Short direction = this.inputPositionReport.getDirection(); final short dir = direction.shortValue(); final short diff = (short)-(dir - 1 + ((dir + 1) / 2)); assert (dir == Constants.EASTBOUND.shortValue() ? diff == 1 : diff == -1); final Integer xway = this.inputPositionReport.getXWay(); final short curSeg = currentSegment.shortValue(); this.segmentToCheck.setXWay(xway); this.segmentToCheck.setDirection(direction); for(int i = 0; i <= 4; ++i) { final short nextSegment = (short)(curSeg + (diff * i)); assert (dir == Constants.EASTBOUND.shortValue() ? nextSegment >= curSeg : nextSegment <= curSeg); this.segmentToCheck.setSegment(new Short(nextSegment)); if(this.previousMinuteAccidents.contains(this.segmentToCheck)) { this.collector.emit(new AccidentNotification(this.inputPositionReport.getTime(), this.inputPositionReport.getTime(), this.segmentToCheck.getSegment(), vid)); break; } } } else { this.inputAccidentTuple.clear(); this.inputAccidentTuple.addAll(input.getValues()); LOGGER.trace(this.inputAccidentTuple.toString()); this.checkMinute(this.inputAccidentTuple.getMinuteNumber()); assert (this.inputAccidentTuple.getMinuteNumber() == this.currentMinute); this.currentMinuteAccidents.add(new SegmentIdentifier(this.inputAccidentTuple)); } this.collector.ack(input); } }
AccidentNotificationBolt extends BaseRichBolt { @Override public void execute(Tuple input) { final String sourceStreamId = input.getSourceStreamId(); if(sourceStreamId.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).shortValue())); } this.collector.ack(input); return; } if(sourceStreamId.equals(TopologyControl.POSITION_REPORTS_STREAM_ID)) { this.inputPositionReport.clear(); this.inputPositionReport.addAll(input.getValues()); LOGGER.trace(this.inputPositionReport.toString()); this.checkMinute(this.inputPositionReport.getMinuteNumber()); if(this.inputPositionReport.isOnExitLane()) { this.collector.ack(input); return; } final Short currentSegment = this.inputPositionReport.getSegment(); final Integer vid = this.inputPositionReport.getVid(); final Short previousSegment = this.allCars.put(vid, currentSegment); if(previousSegment != null && currentSegment.shortValue() == previousSegment.shortValue()) { this.collector.ack(input); return; } final Short direction = this.inputPositionReport.getDirection(); final short dir = direction.shortValue(); final short diff = (short)-(dir - 1 + ((dir + 1) / 2)); assert (dir == Constants.EASTBOUND.shortValue() ? diff == 1 : diff == -1); final Integer xway = this.inputPositionReport.getXWay(); final short curSeg = currentSegment.shortValue(); this.segmentToCheck.setXWay(xway); this.segmentToCheck.setDirection(direction); for(int i = 0; i <= 4; ++i) { final short nextSegment = (short)(curSeg + (diff * i)); assert (dir == Constants.EASTBOUND.shortValue() ? nextSegment >= curSeg : nextSegment <= curSeg); this.segmentToCheck.setSegment(new Short(nextSegment)); if(this.previousMinuteAccidents.contains(this.segmentToCheck)) { this.collector.emit(new AccidentNotification(this.inputPositionReport.getTime(), this.inputPositionReport.getTime(), this.segmentToCheck.getSegment(), vid)); break; } } } else { this.inputAccidentTuple.clear(); this.inputAccidentTuple.addAll(input.getValues()); LOGGER.trace(this.inputAccidentTuple.toString()); this.checkMinute(this.inputAccidentTuple.getMinuteNumber()); assert (this.inputAccidentTuple.getMinuteNumber() == this.currentMinute); this.currentMinuteAccidents.add(new SegmentIdentifier(this.inputAccidentTuple)); } this.collector.ack(input); }  }
AccidentNotificationBolt extends BaseRichBolt { @Override public void execute(Tuple input) { final String sourceStreamId = input.getSourceStreamId(); if(sourceStreamId.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).shortValue())); } this.collector.ack(input); return; } if(sourceStreamId.equals(TopologyControl.POSITION_REPORTS_STREAM_ID)) { this.inputPositionReport.clear(); this.inputPositionReport.addAll(input.getValues()); LOGGER.trace(this.inputPositionReport.toString()); this.checkMinute(this.inputPositionReport.getMinuteNumber()); if(this.inputPositionReport.isOnExitLane()) { this.collector.ack(input); return; } final Short currentSegment = this.inputPositionReport.getSegment(); final Integer vid = this.inputPositionReport.getVid(); final Short previousSegment = this.allCars.put(vid, currentSegment); if(previousSegment != null && currentSegment.shortValue() == previousSegment.shortValue()) { this.collector.ack(input); return; } final Short direction = this.inputPositionReport.getDirection(); final short dir = direction.shortValue(); final short diff = (short)-(dir - 1 + ((dir + 1) / 2)); assert (dir == Constants.EASTBOUND.shortValue() ? diff == 1 : diff == -1); final Integer xway = this.inputPositionReport.getXWay(); final short curSeg = currentSegment.shortValue(); this.segmentToCheck.setXWay(xway); this.segmentToCheck.setDirection(direction); for(int i = 0; i <= 4; ++i) { final short nextSegment = (short)(curSeg + (diff * i)); assert (dir == Constants.EASTBOUND.shortValue() ? nextSegment >= curSeg : nextSegment <= curSeg); this.segmentToCheck.setSegment(new Short(nextSegment)); if(this.previousMinuteAccidents.contains(this.segmentToCheck)) { this.collector.emit(new AccidentNotification(this.inputPositionReport.getTime(), this.inputPositionReport.getTime(), this.segmentToCheck.getSegment(), vid)); break; } } } else { this.inputAccidentTuple.clear(); this.inputAccidentTuple.addAll(input.getValues()); LOGGER.trace(this.inputAccidentTuple.toString()); this.checkMinute(this.inputAccidentTuple.getMinuteNumber()); assert (this.inputAccidentTuple.getMinuteNumber() == this.currentMinute); this.currentMinuteAccidents.add(new SegmentIdentifier(this.inputAccidentTuple)); } 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); }
AccidentNotificationBolt extends BaseRichBolt { @Override public void execute(Tuple input) { final String sourceStreamId = input.getSourceStreamId(); if(sourceStreamId.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).shortValue())); } this.collector.ack(input); return; } if(sourceStreamId.equals(TopologyControl.POSITION_REPORTS_STREAM_ID)) { this.inputPositionReport.clear(); this.inputPositionReport.addAll(input.getValues()); LOGGER.trace(this.inputPositionReport.toString()); this.checkMinute(this.inputPositionReport.getMinuteNumber()); if(this.inputPositionReport.isOnExitLane()) { this.collector.ack(input); return; } final Short currentSegment = this.inputPositionReport.getSegment(); final Integer vid = this.inputPositionReport.getVid(); final Short previousSegment = this.allCars.put(vid, currentSegment); if(previousSegment != null && currentSegment.shortValue() == previousSegment.shortValue()) { this.collector.ack(input); return; } final Short direction = this.inputPositionReport.getDirection(); final short dir = direction.shortValue(); final short diff = (short)-(dir - 1 + ((dir + 1) / 2)); assert (dir == Constants.EASTBOUND.shortValue() ? diff == 1 : diff == -1); final Integer xway = this.inputPositionReport.getXWay(); final short curSeg = currentSegment.shortValue(); this.segmentToCheck.setXWay(xway); this.segmentToCheck.setDirection(direction); for(int i = 0; i <= 4; ++i) { final short nextSegment = (short)(curSeg + (diff * i)); assert (dir == Constants.EASTBOUND.shortValue() ? nextSegment >= curSeg : nextSegment <= curSeg); this.segmentToCheck.setSegment(new Short(nextSegment)); if(this.previousMinuteAccidents.contains(this.segmentToCheck)) { this.collector.emit(new AccidentNotification(this.inputPositionReport.getTime(), this.inputPositionReport.getTime(), this.segmentToCheck.getSegment(), vid)); break; } } } else { this.inputAccidentTuple.clear(); this.inputAccidentTuple.addAll(input.getValues()); LOGGER.trace(this.inputAccidentTuple.toString()); this.checkMinute(this.inputAccidentTuple.getMinuteNumber()); assert (this.inputAccidentTuple.getMinuteNumber() == this.currentMinute); this.currentMinuteAccidents.add(new SegmentIdentifier(this.inputAccidentTuple)); } 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);  }