@Ignore @Test public void testExecute() { GeneralTopologyContext generalContextMock = mock(GeneralTopologyContext.class); when(generalContextMock.getComponentId(anyInt())).thenReturn("componentID"); when(generalContextMock.getComponentOutputFields(anyString(), anyString())).thenReturn( new Fields(TopologyControlOld.ACCOUNT_BALANCE_REQUEST_FIELD_NAME)); AccountBalanceBolt instance = new AccountBalanceBolt(); 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); instance.prepare(new Config(), contextMock, new OutputCollector(collector)); OutputFieldsDeclarer outputFieldsDeclarer = Mockito.mock(OutputFieldsDeclarer.class); instance.declareOutputFields(outputFieldsDeclarer); int queryIdentifier = 1; int expectedTollTime = 1; AccountBalance expResult = new AccountBalance((short)System.currentTimeMillis(), queryIdentifier, 0, expectedTollTime, (short)System.currentTimeMillis()); int vehicleID0 = (int)(random.nextDouble() * 10000); int xWay = 1; int assessedToll = 0; PositionReport posReport0Stopped = EntityHelper.createPosReport(random, vehicleID0, 0, 0 ); AccountBalanceRequest accountBalanceRequest = new AccountBalanceRequest((short)System.currentTimeMillis(), vehicleID0, queryIdentifier); Tuple tuple = new TupleImpl(generalContextMock, new Values(accountBalanceRequest), 1, TopologyControl.TOLL_ASSESSMENTS_STREAM_ID ); instance.execute(tuple); assertEquals(1, collector.acked.size()); AccountBalance result = (AccountBalance)collector.output.get("default").get(0); assertTrue(expResult.getTime() <= result.getTime()); assertEquals(0, result.getBalance()); assertEquals(queryIdentifier, result.getQueryIdentifier()); generalContextMock = mock(GeneralTopologyContext.class); when(generalContextMock.getComponentId(anyInt())).thenReturn("componentID"); when(generalContextMock.getComponentOutputFields(anyString(), anyString())).thenReturn( new Fields(TopologyControlOld.TOLL_NOTIFICATION_FIELD_NAME)); instance = new AccountBalanceBolt(); collector = new TestOutputCollector(); taskMock = new LinkedList<Integer>(); taskMock.add(0); contextMock = mock(TopologyContext.class); when(contextMock.getComponentTasks(anyString())).thenReturn(taskMock); when(contextMock.getThisTaskIndex()).thenReturn(0); instance.prepare(new Config(), contextMock, new OutputCollector(collector)); outputFieldsDeclarer = Mockito.mock(OutputFieldsDeclarer.class); instance.declareOutputFields(outputFieldsDeclarer); tuple = new TupleImpl(generalContextMock, new Values(vehicleID0, xWay, assessedToll, posReport0Stopped), 1, TopologyControl.ACCOUNT_BALANCE_REQUESTS_STREAM_ID ); instance.execute(tuple); assertEquals(1, collector.acked.size()); }
@Override public void execute(Tuple tuple) { if(tuple.getSourceStreamId().equals(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID)) { this.getBalanceAndSend(tuple); } else if(tuple.getSourceStreamId().equals(TopologyControl.ACCOUNT_BALANCE_REQUESTS_STREAM_ID)) { this.updateBalance(tuple); } else { throw new RuntimeException(String.format("Errornous stream subscription. Please report a bug at %s", Helper.ISSUE_REPORT_URL)); } this.collector.ack(tuple); }
AccountBalanceBolt extends BaseRichBolt { @Override public void execute(Tuple tuple) { if(tuple.getSourceStreamId().equals(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID)) { this.getBalanceAndSend(tuple); } else if(tuple.getSourceStreamId().equals(TopologyControl.ACCOUNT_BALANCE_REQUESTS_STREAM_ID)) { this.updateBalance(tuple); } else { throw new RuntimeException(String.format("Errornous stream subscription. Please report a bug at %s", Helper.ISSUE_REPORT_URL)); } this.collector.ack(tuple); } }
AccountBalanceBolt extends BaseRichBolt { @Override public void execute(Tuple tuple) { if(tuple.getSourceStreamId().equals(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID)) { this.getBalanceAndSend(tuple); } else if(tuple.getSourceStreamId().equals(TopologyControl.ACCOUNT_BALANCE_REQUESTS_STREAM_ID)) { this.updateBalance(tuple); } else { throw new RuntimeException(String.format("Errornous stream subscription. Please report a bug at %s", Helper.ISSUE_REPORT_URL)); } this.collector.ack(tuple); }  }
AccountBalanceBolt extends BaseRichBolt { @Override public void execute(Tuple tuple) { if(tuple.getSourceStreamId().equals(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID)) { this.getBalanceAndSend(tuple); } else if(tuple.getSourceStreamId().equals(TopologyControl.ACCOUNT_BALANCE_REQUESTS_STREAM_ID)) { this.updateBalance(tuple); } else { throw new RuntimeException(String.format("Errornous stream subscription. Please report a bug at %s", Helper.ISSUE_REPORT_URL)); } this.collector.ack(tuple); }  @Override void prepare(@SuppressWarnings("rawtypes") Map conf, TopologyContext context, OutputCollector collector); @Override void execute(Tuple tuple); Map<Integer, VehicleAccount> getAllVehicles(); @Override void declareOutputFields(OutputFieldsDeclarer declarer); }
AccountBalanceBolt extends BaseRichBolt { @Override public void execute(Tuple tuple) { if(tuple.getSourceStreamId().equals(TopologyControl.TOLL_ASSESSMENTS_STREAM_ID)) { this.getBalanceAndSend(tuple); } else if(tuple.getSourceStreamId().equals(TopologyControl.ACCOUNT_BALANCE_REQUESTS_STREAM_ID)) { this.updateBalance(tuple); } else { throw new RuntimeException(String.format("Errornous stream subscription. Please report a bug at %s", Helper.ISSUE_REPORT_URL)); } this.collector.ack(tuple); }  @Override void prepare(@SuppressWarnings("rawtypes") Map conf, TopologyContext context, OutputCollector collector); @Override void execute(Tuple tuple); Map<Integer, VehicleAccount> getAllVehicles(); @Override void declareOutputFields(OutputFieldsDeclarer declarer);  }