@Ignore @Test public void testExecute() { GeneralTopologyContext generalContextMock = mock(GeneralTopologyContext.class); when(generalContextMock.getComponentOutputFields(anyString(), anyString())).thenReturn(new Fields("dummy")); when(generalContextMock.getComponentId(anyInt())).thenReturn("componentID"); Fields schema = new Fields(TopologyControlOld.DAILY_EXPEDITURE_REQUEST_FIELD_NAME); when(generalContextMock.getComponentOutputFields(anyString(), anyString())).thenReturn(schema); TestOutputCollector collector = new TestOutputCollector(); List<Integer> taskMock = new LinkedList<Integer>(); taskMock.add(0); TopologyContext contextMock = mock(TopologyContext.class); when(contextMock.getComponentTasks(anyString())).thenReturn(taskMock); when(contextMock.getThisTaskIndex()).thenReturn(0); GeneralTopologyContext context = mock(GeneralTopologyContext.class); when(context.getComponentOutputFields(anyString(), anyString())).thenReturn(new Fields("dummy")); when(context.getComponentId(anyInt())).thenReturn("componentID"); DailyExpenditureBolt instance = new DailyExpenditureBolt(); Config instanceConfig = new Config(); instanceConfig.put(Helper.TOLL_DATA_STORE_CONF_KEY, MemoryTollDataStore.class.getName()); instance.prepare(instanceConfig, contextMock, new OutputCollector(collector)); OutputFieldsDeclarer outputFieldsDeclarer = Mockito.mock(OutputFieldsDeclarer.class); instance.declareOutputFields(outputFieldsDeclarer); int vehicleIdentifierValid = 1; int xWay = 1; int queryIdentifier = 2; short day0 = 1; int expectedToll = 4748; instance.getDataStore().storeToll(xWay, day0, vehicleIdentifierValid, expectedToll); DailyExpenditureRequest dailyExpenditureRequest = new DailyExpenditureRequest( (short)System.currentTimeMillis(), vehicleIdentifierValid, xWay, queryIdentifier, day0); Tuple tuple = new TupleImpl(generalContextMock, new Values(dailyExpenditureRequest), 1, null ); instance.execute(tuple); assertEquals(1, collector.acked.size()); assertEquals(1, collector.output.size()); assertEquals(1, collector.output.get(Utils.DEFAULT_STREAM_ID).size()); List<Object> resultTuple1 = collector.output.get(Utils.DEFAULT_STREAM_ID).get(0); assertEquals(AbstractLRBTuple.DAILY_EXPENDITURE_REQUEST, resultTuple1.get(0)); assertEquals(dailyExpenditureRequest.getTime(), resultTuple1.get(1)); assertEquals(queryIdentifier, resultTuple1.get(3)); assertEquals(expectedToll, resultTuple1.get(4)); int vehicleIdentifierInvalid = 2; dailyExpenditureRequest = new DailyExpenditureRequest(day0, vehicleIdentifierInvalid, xWay, queryIdentifier, day0); tuple = new TupleImpl(generalContextMock, new Values(dailyExpenditureRequest), 1, null ); instance.execute(tuple); assertEquals(2, collector.acked.size()); assertEquals(1, collector.output.size()); assertEquals(2, collector.output.get(Utils.DEFAULT_STREAM_ID).size()); List<Object> resultTuple2 = collector.output.get(Utils.DEFAULT_STREAM_ID).get(1); assertEquals(AbstractLRBTuple.DAILY_EXPENDITURE_REQUEST, resultTuple2.get(0)); assertEquals(dailyExpenditureRequest.getTime(), resultTuple2.get(1)); assertEquals(queryIdentifier, resultTuple2.get(3)); assertEquals(20, resultTuple2.get(4)); }
@Override public void execute(Tuple tuple) { Fields fields = tuple.getFields(); if(fields.contains(TopologyControlOld.DAILY_EXPEDITURE_REQUEST_FIELD_NAME)) { DailyExpenditureRequest exp = (DailyExpenditureRequest)tuple .getValueByField(TopologyControlOld.DAILY_EXPEDITURE_REQUEST_FIELD_NAME); int vehicleIdentifier = exp.getVid(); Values values; Integer toll = this.dataStore.retrieveToll(exp.getXWay(), exp.getDay(), vehicleIdentifier); if(toll != null) { LOG.debug("ExpenditureRequest: found vehicle identifier %d", vehicleIdentifier); values = new Values(AbstractLRBTuple.DAILY_EXPENDITURE_REQUEST, exp.getTime(), exp.getQid(), toll); } else { values = new Values(AbstractLRBTuple.DAILY_EXPENDITURE_REQUEST, exp.getTime(), exp.getQid(), 20); } this.collector.emit(values); } this.collector.ack(tuple); }
DailyExpenditureBolt extends BaseRichBolt { @Override public void execute(Tuple tuple) { Fields fields = tuple.getFields(); if(fields.contains(TopologyControlOld.DAILY_EXPEDITURE_REQUEST_FIELD_NAME)) { DailyExpenditureRequest exp = (DailyExpenditureRequest)tuple .getValueByField(TopologyControlOld.DAILY_EXPEDITURE_REQUEST_FIELD_NAME); int vehicleIdentifier = exp.getVid(); Values values; Integer toll = this.dataStore.retrieveToll(exp.getXWay(), exp.getDay(), vehicleIdentifier); if(toll != null) { LOG.debug("ExpenditureRequest: found vehicle identifier %d", vehicleIdentifier); values = new Values(AbstractLRBTuple.DAILY_EXPENDITURE_REQUEST, exp.getTime(), exp.getQid(), toll); } else { values = new Values(AbstractLRBTuple.DAILY_EXPENDITURE_REQUEST, exp.getTime(), exp.getQid(), 20); } this.collector.emit(values); } this.collector.ack(tuple); } }
DailyExpenditureBolt extends BaseRichBolt { @Override public void execute(Tuple tuple) { Fields fields = tuple.getFields(); if(fields.contains(TopologyControlOld.DAILY_EXPEDITURE_REQUEST_FIELD_NAME)) { DailyExpenditureRequest exp = (DailyExpenditureRequest)tuple .getValueByField(TopologyControlOld.DAILY_EXPEDITURE_REQUEST_FIELD_NAME); int vehicleIdentifier = exp.getVid(); Values values; Integer toll = this.dataStore.retrieveToll(exp.getXWay(), exp.getDay(), vehicleIdentifier); if(toll != null) { LOG.debug("ExpenditureRequest: found vehicle identifier %d", vehicleIdentifier); values = new Values(AbstractLRBTuple.DAILY_EXPENDITURE_REQUEST, exp.getTime(), exp.getQid(), toll); } else { values = new Values(AbstractLRBTuple.DAILY_EXPENDITURE_REQUEST, exp.getTime(), exp.getQid(), 20); } this.collector.emit(values); } this.collector.ack(tuple); } DailyExpenditureBolt(); }
DailyExpenditureBolt extends BaseRichBolt { @Override public void execute(Tuple tuple) { Fields fields = tuple.getFields(); if(fields.contains(TopologyControlOld.DAILY_EXPEDITURE_REQUEST_FIELD_NAME)) { DailyExpenditureRequest exp = (DailyExpenditureRequest)tuple .getValueByField(TopologyControlOld.DAILY_EXPEDITURE_REQUEST_FIELD_NAME); int vehicleIdentifier = exp.getVid(); Values values; Integer toll = this.dataStore.retrieveToll(exp.getXWay(), exp.getDay(), vehicleIdentifier); if(toll != null) { LOG.debug("ExpenditureRequest: found vehicle identifier %d", vehicleIdentifier); values = new Values(AbstractLRBTuple.DAILY_EXPENDITURE_REQUEST, exp.getTime(), exp.getQid(), toll); } else { values = new Values(AbstractLRBTuple.DAILY_EXPENDITURE_REQUEST, exp.getTime(), exp.getQid(), 20); } this.collector.emit(values); } this.collector.ack(tuple); } DailyExpenditureBolt(); TollDataStore getDataStore(); @Override void prepare(@SuppressWarnings("rawtypes") Map conf, TopologyContext context, OutputCollector collector); @Override void execute(Tuple tuple); @Override void declareOutputFields(OutputFieldsDeclarer declarer); }
DailyExpenditureBolt extends BaseRichBolt { @Override public void execute(Tuple tuple) { Fields fields = tuple.getFields(); if(fields.contains(TopologyControlOld.DAILY_EXPEDITURE_REQUEST_FIELD_NAME)) { DailyExpenditureRequest exp = (DailyExpenditureRequest)tuple .getValueByField(TopologyControlOld.DAILY_EXPEDITURE_REQUEST_FIELD_NAME); int vehicleIdentifier = exp.getVid(); Values values; Integer toll = this.dataStore.retrieveToll(exp.getXWay(), exp.getDay(), vehicleIdentifier); if(toll != null) { LOG.debug("ExpenditureRequest: found vehicle identifier %d", vehicleIdentifier); values = new Values(AbstractLRBTuple.DAILY_EXPENDITURE_REQUEST, exp.getTime(), exp.getQid(), toll); } else { values = new Values(AbstractLRBTuple.DAILY_EXPENDITURE_REQUEST, exp.getTime(), exp.getQid(), 20); } this.collector.emit(values); } this.collector.ack(tuple); } DailyExpenditureBolt(); TollDataStore getDataStore(); @Override void prepare(@SuppressWarnings("rawtypes") Map conf, TopologyContext context, OutputCollector collector); @Override void execute(Tuple tuple); @Override void declareOutputFields(OutputFieldsDeclarer declarer);  }