@SuppressWarnings("boxing") @Test public void testExecute() { Tuple tuple = mock(Tuple.class); OngoingStubbing<List<Object>> valueStub = when(tuple.getValues()); List<TollNotification> expectedNotifications = new ArrayList<TollNotification>(); List<TollNotification> expectedAssessments = new ArrayList<TollNotification>(); List<String> streamIdsToBeMocked = new LinkedList<String>(); short time; time = (short)0; streamIdsToBeMocked.add(TopologyControl.POSITION_REPORTS_STREAM_ID); valueStub = valueStub.thenReturn(new PositionReport(time, vid, spd, xway, l0, d, (short)5, pos)); expectedNotifications.add(new TollNotification(time, time, vid, -1, 0)); time = (short)30; streamIdsToBeMocked.add(TopologyControl.POSITION_REPORTS_STREAM_ID); valueStub = valueStub.thenReturn(new PositionReport(time, vid, spd, xway, l1, d, (short)5, pos)); streamIdsToBeMocked.add(TopologyControl.CAR_COUNTS_STREAM_ID); valueStub = valueStub.thenReturn(new CountTuple((short)59, xway, (short)6, d, 50)); time = (short)60; streamIdsToBeMocked.add(TopologyControl.POSITION_REPORTS_STREAM_ID); valueStub = valueStub.thenReturn(new PositionReport(time, vid, spd, xway, l1, d, (short)6, pos)); expectedNotifications.add(new TollNotification(time, time, vid, -1, 0)); time = (short)90; streamIdsToBeMocked.add(TopologyControl.POSITION_REPORTS_STREAM_ID); valueStub = valueStub.thenReturn(new PositionReport(time, vid, spd, xway, l2, d, (short)7, pos)); expectedNotifications.add(new TollNotification(time, time, vid, -1, 0)); time = (short)120; streamIdsToBeMocked.add(TopologyControl.POSITION_REPORTS_STREAM_ID); valueStub = valueStub.thenReturn(new PositionReport(time, vid, spd, xway, l4, d, (short)8, pos)); streamIdsToBeMocked.add(TopologyControl.CAR_COUNTS_STREAM_ID); valueStub = valueStub.thenReturn(new CountTuple((short)179, xway, (short)10, d, 51)); time = (short)200; streamIdsToBeMocked.add(TopologyControl.POSITION_REPORTS_STREAM_ID); valueStub = valueStub.thenReturn(new PositionReport(time, vid, spd, xway, l0, d, (short)10, pos)); expectedNotifications.add(new TollNotification(time, time, vid, -1, 2)); time = (short)230; streamIdsToBeMocked.add(TopologyControl.POSITION_REPORTS_STREAM_ID); valueStub = valueStub.thenReturn(new PositionReport(time, vid, spd, xway, l1, d, (short)10, pos)); streamIdsToBeMocked.add(TopologyControl.CAR_COUNTS_STREAM_ID); valueStub = valueStub.thenReturn(new CountTuple((short)239, xway, (short)11, d, 52)); streamIdsToBeMocked.add(TopologyControl.LAVS_STREAM_ID); valueStub = valueStub.thenReturn(new LavTuple((short)299, xway, (short)11, d, 39)); streamIdsToBeMocked.add(TopologyControl.ACCIDENTS_STREAM_ID); valueStub = valueStub.thenReturn(new AccidentTuple((short)239, xway, (short)10, d)); streamIdsToBeMocked.add(TopologyControl.ACCIDENTS_STREAM_ID); valueStub = valueStub.thenReturn(new AccidentTuple((short)239, xway, (short)16, d)); time = (short)260; streamIdsToBeMocked.add(TopologyControl.POSITION_REPORTS_STREAM_ID); valueStub = valueStub.thenReturn(new PositionReport(time, vid, spd, xway, l1, d, (short)11, pos)); expectedNotifications.add(new TollNotification(time, time, vid, 39, 8)); expectedAssessments.add(new TollNotification((short)200, (short)200, vid, -1, 2)); time = (short)290; streamIdsToBeMocked.add(TopologyControl.POSITION_REPORTS_STREAM_ID); valueStub = valueStub.thenReturn(new PositionReport(time, vid, spd, xway, l4, d, (short)12, pos)); expectedAssessments.add(new TollNotification((short)260, (short)260, vid, 39, 8)); streamIdsToBeMocked.add(TopologyControl.CAR_COUNTS_STREAM_ID); valueStub = valueStub.thenReturn(new CountTuple((short)359, xway, (short)20, d, 51)); streamIdsToBeMocked.add(TopologyControl.LAVS_STREAM_ID); valueStub = valueStub.thenReturn(new LavTuple((short)419, xway, (short)20, d, 40)); time = (short)400; streamIdsToBeMocked.add(TopologyControl.POSITION_REPORTS_STREAM_ID); valueStub = valueStub.thenReturn(new PositionReport(time, vid, spd, xway, l0, d, (short)20, pos)); expectedNotifications.add(new TollNotification(time, time, vid, 40, 0)); streamIdsToBeMocked.add(TopologyControl.CAR_COUNTS_STREAM_ID); valueStub = valueStub.thenReturn(new CountTuple((short)479, xway, (short)30, d, 51)); time = (short)500; streamIdsToBeMocked.add(TopologyControl.POSITION_REPORTS_STREAM_ID); valueStub = valueStub.thenReturn(new PositionReport(time, vid, spd, xway, l0, d, (short)30, pos)); expectedNotifications.add(new TollNotification(time, time, vid, -1, 2)); time = (short)530; streamIdsToBeMocked.add(TopologyControl.POSITION_REPORTS_STREAM_ID); valueStub = valueStub.thenReturn(new PositionReport(time, vid, spd, xway, l1, d, (short)30, pos)); streamIdsToBeMocked.add(TopologyControl.CAR_COUNTS_STREAM_ID); valueStub = valueStub.thenReturn(new CountTuple((short)539, xway, (short)31, d, 52)); streamIdsToBeMocked.add(TopologyControl.LAVS_STREAM_ID); valueStub = valueStub.thenReturn(new LavTuple((short)599, xway, (short)31, d, 39)); streamIdsToBeMocked.add(TopologyControl.ACCIDENTS_STREAM_ID); valueStub = valueStub.thenReturn(new AccidentTuple((short)539, xway, (short)31, d)); time = (short)560; streamIdsToBeMocked.add(TopologyControl.POSITION_REPORTS_STREAM_ID); valueStub = valueStub.thenReturn(new PositionReport(time, vid, spd, xway, l1, d, (short)31, pos)); expectedNotifications.add(new TollNotification(time, time, vid, 39, 0)); expectedAssessments.add(new TollNotification((short)500, (short)500, vid, -1, 2)); time = (short)590; streamIdsToBeMocked.add(TopologyControl.POSITION_REPORTS_STREAM_ID); valueStub = valueStub.thenReturn(new PositionReport(time, vid, spd, xway, l4, d, (short)32, pos)); streamIdsToBeMocked.add(TopologyControl.CAR_COUNTS_STREAM_ID); valueStub = valueStub.thenReturn(new CountTuple((short)599, xway, (short)30, d, 51)); time = (short)600; streamIdsToBeMocked.add(TopologyControl.POSITION_REPORTS_STREAM_ID); valueStub = valueStub.thenReturn(new PositionReport(time, vid, spd, xway, l0, d, (short)30, pos)); expectedNotifications.add(new TollNotification(time, time, vid, -1, 2)); time = (short)630; streamIdsToBeMocked.add(TopologyControl.POSITION_REPORTS_STREAM_ID); valueStub = valueStub.thenReturn(new PositionReport(time, vid, spd, xway, l1, d, (short)30, pos)); streamIdsToBeMocked.add(TopologyControl.CAR_COUNTS_STREAM_ID); valueStub = valueStub.thenReturn(new CountTuple((short)659, xway, (short)31, d, 52)); streamIdsToBeMocked.add(TopologyControl.LAVS_STREAM_ID); valueStub = valueStub.thenReturn(new LavTuple((short)719, xway, (short)31, d, 39)); streamIdsToBeMocked.add(TopologyControl.ACCIDENTS_STREAM_ID); valueStub = valueStub.thenReturn(new AccidentTuple((short)659, xway, (short)35, d)); time = (short)660; streamIdsToBeMocked.add(TopologyControl.POSITION_REPORTS_STREAM_ID); valueStub = valueStub.thenReturn(new PositionReport(time, vid, spd, xway, l1, d, (short)31, pos)); expectedNotifications.add(new TollNotification(time, time, vid, 39, 0)); expectedAssessments.add(new TollNotification((short)600, (short)600, vid, -1, 2)); time = (short)690; streamIdsToBeMocked.add(TopologyControl.POSITION_REPORTS_STREAM_ID); valueStub = valueStub.thenReturn(new PositionReport(time, vid, spd, xway, l4, d, (short)32, pos)); valueStub.thenReturn(null); OngoingStubbing<String> streamStub = when(tuple.getSourceStreamId()); for(String streamId : streamIdsToBeMocked) { streamStub = streamStub.thenReturn(streamId); } streamStub.thenReturn(null); TollNotificationBolt bolt = new TollNotificationBolt(); TestOutputCollector collector = new TestOutputCollector(); bolt.prepare(null, null, new OutputCollector(collector)); for(int i = 0; i < streamIdsToBeMocked.size(); ++i) { bolt.execute(tuple); } Assert.assertEquals(2, collector.output.size()); Assert.assertEquals(expectedNotifications, collector.output.get(TopologyControl.TOLL_NOTIFICATIONS_STREAM_ID)); Assert.assertEquals(expectedAssessments, collector.output.get(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID)); 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(3, collector.output.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)); }
@Override public void execute(Tuple input) { final String inputStreamId = input.getSourceStreamId(); if(inputStreamId.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(inputStreamId.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()) { final TollNotification lastNotification = this.lastTollNotification.remove(this.inputPositionReport .getVid()); if(lastNotification != null) { this.collector.emit(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID, lastNotification); } 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; } int toll = 0; Integer lav = this.previousMinuteLavs.get(new SegmentIdentifier(this.inputPositionReport)); final int lavValue; if(lav != null) { lavValue = lav.intValue(); } else { lav = new Integer(-1); lavValue = -1; } if(lavValue < 40) { final Integer count = this.previousMinuteCounts.get(new SegmentIdentifier(this.inputPositionReport)); int carCount = 0; if(count != null) { carCount = count.intValue(); } if(carCount > 50) { 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); int i; for(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)) { break; } } if(i == 5) { final int var = carCount - 50; toll = 2 * var * var; } } } final TollNotification tollNotification = new TollNotification(this.inputPositionReport.getTime(), this.inputPositionReport.getTime(), vid, lav, new Integer(toll)); final TollNotification lastNotification; if(toll != 0) { lastNotification = this.lastTollNotification.put(vid, new TollNotification(tollNotification.getTime(), tollNotification.getEmit(), tollNotification.getVid(), tollNotification.getSpeed(), tollNotification.getToll())); } else { lastNotification = this.lastTollNotification.remove(vid); } if(lastNotification != null) { this.collector.emit(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID, lastNotification); } this.collector.emit(TopologyControl.TOLL_NOTIFICATIONS_STREAM_ID, tollNotification); } else if(inputStreamId.equals(TopologyControl.ACCIDENTS_STREAM_ID)) { 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)); } else if(inputStreamId.equals(TopologyControl.CAR_COUNTS_STREAM_ID)) { this.inputCountTuple.clear(); this.inputCountTuple.addAll(input.getValues()); LOGGER.trace(this.inputCountTuple.toString()); this.checkMinute(this.inputCountTuple.getMinuteNumber()); assert (this.inputCountTuple.getMinuteNumber() == this.currentMinute); this.currentMinuteCounts.put(new SegmentIdentifier(this.inputCountTuple), this.inputCountTuple.getCount()); } else if(inputStreamId.equals(TopologyControl.LAVS_STREAM_ID)) { this.inputLavTuple.clear(); this.inputLavTuple.addAll(input.getValues()); LOGGER.trace(this.inputLavTuple.toString()); this.checkMinute((short)(this.inputLavTuple.getMinuteNumber() - 1)); assert (this.inputLavTuple.getMinuteNumber() - 1 == this.currentMinute); this.currentMinuteLavs.put(new SegmentIdentifier(this.inputLavTuple), this.inputLavTuple.getLav()); } else { LOGGER.error("Unknown input stream: '" + inputStreamId + "' for tuple " + input); throw new RuntimeException("Unknown input stream: '" + inputStreamId + "' for tuple " + input); } this.collector.ack(input); }
TollNotificationBolt extends BaseRichBolt { @Override public void execute(Tuple input) { final String inputStreamId = input.getSourceStreamId(); if(inputStreamId.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(inputStreamId.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()) { final TollNotification lastNotification = this.lastTollNotification.remove(this.inputPositionReport .getVid()); if(lastNotification != null) { this.collector.emit(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID, lastNotification); } 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; } int toll = 0; Integer lav = this.previousMinuteLavs.get(new SegmentIdentifier(this.inputPositionReport)); final int lavValue; if(lav != null) { lavValue = lav.intValue(); } else { lav = new Integer(-1); lavValue = -1; } if(lavValue < 40) { final Integer count = this.previousMinuteCounts.get(new SegmentIdentifier(this.inputPositionReport)); int carCount = 0; if(count != null) { carCount = count.intValue(); } if(carCount > 50) { 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); int i; for(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)) { break; } } if(i == 5) { final int var = carCount - 50; toll = 2 * var * var; } } } final TollNotification tollNotification = new TollNotification(this.inputPositionReport.getTime(), this.inputPositionReport.getTime(), vid, lav, new Integer(toll)); final TollNotification lastNotification; if(toll != 0) { lastNotification = this.lastTollNotification.put(vid, new TollNotification(tollNotification.getTime(), tollNotification.getEmit(), tollNotification.getVid(), tollNotification.getSpeed(), tollNotification.getToll())); } else { lastNotification = this.lastTollNotification.remove(vid); } if(lastNotification != null) { this.collector.emit(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID, lastNotification); } this.collector.emit(TopologyControl.TOLL_NOTIFICATIONS_STREAM_ID, tollNotification); } else if(inputStreamId.equals(TopologyControl.ACCIDENTS_STREAM_ID)) { 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)); } else if(inputStreamId.equals(TopologyControl.CAR_COUNTS_STREAM_ID)) { this.inputCountTuple.clear(); this.inputCountTuple.addAll(input.getValues()); LOGGER.trace(this.inputCountTuple.toString()); this.checkMinute(this.inputCountTuple.getMinuteNumber()); assert (this.inputCountTuple.getMinuteNumber() == this.currentMinute); this.currentMinuteCounts.put(new SegmentIdentifier(this.inputCountTuple), this.inputCountTuple.getCount()); } else if(inputStreamId.equals(TopologyControl.LAVS_STREAM_ID)) { this.inputLavTuple.clear(); this.inputLavTuple.addAll(input.getValues()); LOGGER.trace(this.inputLavTuple.toString()); this.checkMinute((short)(this.inputLavTuple.getMinuteNumber() - 1)); assert (this.inputLavTuple.getMinuteNumber() - 1 == this.currentMinute); this.currentMinuteLavs.put(new SegmentIdentifier(this.inputLavTuple), this.inputLavTuple.getLav()); } else { LOGGER.error("Unknown input stream: '" + inputStreamId + "' for tuple " + input); throw new RuntimeException("Unknown input stream: '" + inputStreamId + "' for tuple " + input); } this.collector.ack(input); } }
TollNotificationBolt extends BaseRichBolt { @Override public void execute(Tuple input) { final String inputStreamId = input.getSourceStreamId(); if(inputStreamId.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(inputStreamId.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()) { final TollNotification lastNotification = this.lastTollNotification.remove(this.inputPositionReport .getVid()); if(lastNotification != null) { this.collector.emit(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID, lastNotification); } 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; } int toll = 0; Integer lav = this.previousMinuteLavs.get(new SegmentIdentifier(this.inputPositionReport)); final int lavValue; if(lav != null) { lavValue = lav.intValue(); } else { lav = new Integer(-1); lavValue = -1; } if(lavValue < 40) { final Integer count = this.previousMinuteCounts.get(new SegmentIdentifier(this.inputPositionReport)); int carCount = 0; if(count != null) { carCount = count.intValue(); } if(carCount > 50) { 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); int i; for(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)) { break; } } if(i == 5) { final int var = carCount - 50; toll = 2 * var * var; } } } final TollNotification tollNotification = new TollNotification(this.inputPositionReport.getTime(), this.inputPositionReport.getTime(), vid, lav, new Integer(toll)); final TollNotification lastNotification; if(toll != 0) { lastNotification = this.lastTollNotification.put(vid, new TollNotification(tollNotification.getTime(), tollNotification.getEmit(), tollNotification.getVid(), tollNotification.getSpeed(), tollNotification.getToll())); } else { lastNotification = this.lastTollNotification.remove(vid); } if(lastNotification != null) { this.collector.emit(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID, lastNotification); } this.collector.emit(TopologyControl.TOLL_NOTIFICATIONS_STREAM_ID, tollNotification); } else if(inputStreamId.equals(TopologyControl.ACCIDENTS_STREAM_ID)) { 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)); } else if(inputStreamId.equals(TopologyControl.CAR_COUNTS_STREAM_ID)) { this.inputCountTuple.clear(); this.inputCountTuple.addAll(input.getValues()); LOGGER.trace(this.inputCountTuple.toString()); this.checkMinute(this.inputCountTuple.getMinuteNumber()); assert (this.inputCountTuple.getMinuteNumber() == this.currentMinute); this.currentMinuteCounts.put(new SegmentIdentifier(this.inputCountTuple), this.inputCountTuple.getCount()); } else if(inputStreamId.equals(TopologyControl.LAVS_STREAM_ID)) { this.inputLavTuple.clear(); this.inputLavTuple.addAll(input.getValues()); LOGGER.trace(this.inputLavTuple.toString()); this.checkMinute((short)(this.inputLavTuple.getMinuteNumber() - 1)); assert (this.inputLavTuple.getMinuteNumber() - 1 == this.currentMinute); this.currentMinuteLavs.put(new SegmentIdentifier(this.inputLavTuple), this.inputLavTuple.getLav()); } else { LOGGER.error("Unknown input stream: '" + inputStreamId + "' for tuple " + input); throw new RuntimeException("Unknown input stream: '" + inputStreamId + "' for tuple " + input); } this.collector.ack(input); }  }
TollNotificationBolt extends BaseRichBolt { @Override public void execute(Tuple input) { final String inputStreamId = input.getSourceStreamId(); if(inputStreamId.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(inputStreamId.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()) { final TollNotification lastNotification = this.lastTollNotification.remove(this.inputPositionReport .getVid()); if(lastNotification != null) { this.collector.emit(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID, lastNotification); } 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; } int toll = 0; Integer lav = this.previousMinuteLavs.get(new SegmentIdentifier(this.inputPositionReport)); final int lavValue; if(lav != null) { lavValue = lav.intValue(); } else { lav = new Integer(-1); lavValue = -1; } if(lavValue < 40) { final Integer count = this.previousMinuteCounts.get(new SegmentIdentifier(this.inputPositionReport)); int carCount = 0; if(count != null) { carCount = count.intValue(); } if(carCount > 50) { 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); int i; for(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)) { break; } } if(i == 5) { final int var = carCount - 50; toll = 2 * var * var; } } } final TollNotification tollNotification = new TollNotification(this.inputPositionReport.getTime(), this.inputPositionReport.getTime(), vid, lav, new Integer(toll)); final TollNotification lastNotification; if(toll != 0) { lastNotification = this.lastTollNotification.put(vid, new TollNotification(tollNotification.getTime(), tollNotification.getEmit(), tollNotification.getVid(), tollNotification.getSpeed(), tollNotification.getToll())); } else { lastNotification = this.lastTollNotification.remove(vid); } if(lastNotification != null) { this.collector.emit(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID, lastNotification); } this.collector.emit(TopologyControl.TOLL_NOTIFICATIONS_STREAM_ID, tollNotification); } else if(inputStreamId.equals(TopologyControl.ACCIDENTS_STREAM_ID)) { 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)); } else if(inputStreamId.equals(TopologyControl.CAR_COUNTS_STREAM_ID)) { this.inputCountTuple.clear(); this.inputCountTuple.addAll(input.getValues()); LOGGER.trace(this.inputCountTuple.toString()); this.checkMinute(this.inputCountTuple.getMinuteNumber()); assert (this.inputCountTuple.getMinuteNumber() == this.currentMinute); this.currentMinuteCounts.put(new SegmentIdentifier(this.inputCountTuple), this.inputCountTuple.getCount()); } else if(inputStreamId.equals(TopologyControl.LAVS_STREAM_ID)) { this.inputLavTuple.clear(); this.inputLavTuple.addAll(input.getValues()); LOGGER.trace(this.inputLavTuple.toString()); this.checkMinute((short)(this.inputLavTuple.getMinuteNumber() - 1)); assert (this.inputLavTuple.getMinuteNumber() - 1 == this.currentMinute); this.currentMinuteLavs.put(new SegmentIdentifier(this.inputLavTuple), this.inputLavTuple.getLav()); } else { LOGGER.error("Unknown input stream: '" + inputStreamId + "' for tuple " + input); throw new RuntimeException("Unknown input stream: '" + inputStreamId + "' for tuple " + input); } 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); }
TollNotificationBolt extends BaseRichBolt { @Override public void execute(Tuple input) { final String inputStreamId = input.getSourceStreamId(); if(inputStreamId.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(inputStreamId.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()) { final TollNotification lastNotification = this.lastTollNotification.remove(this.inputPositionReport .getVid()); if(lastNotification != null) { this.collector.emit(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID, lastNotification); } 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; } int toll = 0; Integer lav = this.previousMinuteLavs.get(new SegmentIdentifier(this.inputPositionReport)); final int lavValue; if(lav != null) { lavValue = lav.intValue(); } else { lav = new Integer(-1); lavValue = -1; } if(lavValue < 40) { final Integer count = this.previousMinuteCounts.get(new SegmentIdentifier(this.inputPositionReport)); int carCount = 0; if(count != null) { carCount = count.intValue(); } if(carCount > 50) { 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); int i; for(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)) { break; } } if(i == 5) { final int var = carCount - 50; toll = 2 * var * var; } } } final TollNotification tollNotification = new TollNotification(this.inputPositionReport.getTime(), this.inputPositionReport.getTime(), vid, lav, new Integer(toll)); final TollNotification lastNotification; if(toll != 0) { lastNotification = this.lastTollNotification.put(vid, new TollNotification(tollNotification.getTime(), tollNotification.getEmit(), tollNotification.getVid(), tollNotification.getSpeed(), tollNotification.getToll())); } else { lastNotification = this.lastTollNotification.remove(vid); } if(lastNotification != null) { this.collector.emit(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID, lastNotification); } this.collector.emit(TopologyControl.TOLL_NOTIFICATIONS_STREAM_ID, tollNotification); } else if(inputStreamId.equals(TopologyControl.ACCIDENTS_STREAM_ID)) { 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)); } else if(inputStreamId.equals(TopologyControl.CAR_COUNTS_STREAM_ID)) { this.inputCountTuple.clear(); this.inputCountTuple.addAll(input.getValues()); LOGGER.trace(this.inputCountTuple.toString()); this.checkMinute(this.inputCountTuple.getMinuteNumber()); assert (this.inputCountTuple.getMinuteNumber() == this.currentMinute); this.currentMinuteCounts.put(new SegmentIdentifier(this.inputCountTuple), this.inputCountTuple.getCount()); } else if(inputStreamId.equals(TopologyControl.LAVS_STREAM_ID)) { this.inputLavTuple.clear(); this.inputLavTuple.addAll(input.getValues()); LOGGER.trace(this.inputLavTuple.toString()); this.checkMinute((short)(this.inputLavTuple.getMinuteNumber() - 1)); assert (this.inputLavTuple.getMinuteNumber() - 1 == this.currentMinute); this.currentMinuteLavs.put(new SegmentIdentifier(this.inputLavTuple), this.inputLavTuple.getLav()); } else { LOGGER.error("Unknown input stream: '" + inputStreamId + "' for tuple " + input); throw new RuntimeException("Unknown input stream: '" + inputStreamId + "' for tuple " + input); } 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);  }