@Test public void testExecute() { ForwardBolt bolt = new ForwardBolt(new Fields("dummy")); TestOutputCollector collector = new TestOutputCollector(); bolt.prepare(null, null, new OutputCollector(collector)); LinkedList<Tuple> tuples = new LinkedList<Tuple>(); List<List<Object>> result = new LinkedList<List<Object>>(); for(int i = 0; i < 3; ++i) { ArrayList<Object> attributes = new ArrayList<Object>(); attributes.add(new Integer(i)); tuples.add(mock(Tuple.class)); when(tuples.get(i).getValues()).thenReturn(attributes); result.add(attributes); bolt.execute(tuples.get(i)); Assert.assertEquals(tuples, collector.acked); } Assert.assertEquals(result, collector.output.get(Utils.DEFAULT_STREAM_ID)); }
@Override public void execute(Tuple input) { for(String streamId : this.outputStreams) { List<Integer> receiverIds = this.collector.emit(streamId, input.getValues()); logger.trace("forwarded tuple {} to output stream {} to receiver tasks with IDs {}", input, streamId, receiverIds); } this.collector.ack(input); }
ForwardBolt implements IRichBolt { @Override public void execute(Tuple input) { for(String streamId : this.outputStreams) { List<Integer> receiverIds = this.collector.emit(streamId, input.getValues()); logger.trace("forwarded tuple {} to output stream {} to receiver tasks with IDs {}", input, streamId, receiverIds); } this.collector.ack(input); } }
ForwardBolt implements IRichBolt { @Override public void execute(Tuple input) { for(String streamId : this.outputStreams) { List<Integer> receiverIds = this.collector.emit(streamId, input.getValues()); logger.trace("forwarded tuple {} to output stream {} to receiver tasks with IDs {}", input, streamId, receiverIds); } this.collector.ack(input); } ForwardBolt();  ForwardBolt(Fields schema);  ForwardBolt(Fields schema, String[] outputStreamIds); }
ForwardBolt implements IRichBolt { @Override public void execute(Tuple input) { for(String streamId : this.outputStreams) { List<Integer> receiverIds = this.collector.emit(streamId, input.getValues()); logger.trace("forwarded tuple {} to output stream {} to receiver tasks with IDs {}", input, streamId, receiverIds); } this.collector.ack(input); } ForwardBolt();  ForwardBolt(Fields schema);  ForwardBolt(Fields schema, String[] outputStreamIds); @Override void prepare(@SuppressWarnings("rawtypes") Map stormConf, TopologyContext context, OutputCollector collector); @Override void execute(Tuple input); @Override void cleanup(); @Override void declareOutputFields(OutputFieldsDeclarer declarer); @Override Map<String, Object> getComponentConfiguration(); }
ForwardBolt implements IRichBolt { @Override public void execute(Tuple input) { for(String streamId : this.outputStreams) { List<Integer> receiverIds = this.collector.emit(streamId, input.getValues()); logger.trace("forwarded tuple {} to output stream {} to receiver tasks with IDs {}", input, streamId, receiverIds); } this.collector.ack(input); } ForwardBolt();  ForwardBolt(Fields schema);  ForwardBolt(Fields schema, String[] outputStreamIds); @Override void prepare(@SuppressWarnings("rawtypes") Map stormConf, TopologyContext context, OutputCollector collector); @Override void execute(Tuple input); @Override void cleanup(); @Override void declareOutputFields(OutputFieldsDeclarer declarer); @Override Map<String, Object> getComponentConfiguration();  }