@Test(timeout = 1000) public void testNextTuple() { final int batchSize = 2 + this.r.nextInt(9); final String streamId = Utils.DEFAULT_STREAM_ID; final String sourceId = "sourceId"; Grouping groupingMock = mock(Grouping.class); HashMap<String, Map<String, Grouping>> targets = new HashMap<String, Map<String, Grouping>>(); Map<String, Grouping> receiver = new HashMap<String, Grouping>(); receiver.put("receiverId", groupingMock); targets.put(streamId, receiver); TopologyContext context = mock(TopologyContext.class); when(context.getThisComponentId()).thenReturn(sourceId); when(context.getThisTargets()).thenReturn(targets); when(context.getComponentOutputFields(sourceId, streamId)).thenReturn(new Fields("dummy")); RandomSpout userSpout = new RandomSpout(1, 1000, new String[] {Utils.DEFAULT_STREAM_ID}, this.seed); SpoutOutputBatcher batcher = new SpoutOutputBatcher(userSpout, batchSize); TestSpoutOutputCollector collector = new TestSpoutOutputCollector(); batcher.open(null, context, new SpoutOutputCollector(collector)); batcher.nextTuple(); Assert.assertEquals(1, collector.output.get(Utils.DEFAULT_STREAM_ID).size()); }
@Override public void nextTuple() { while(true) { this.batchCollector.tupleEmitted = false; this.batchCollector.batchEmitted = false; this.wrappedSpout.nextTuple(); if(!this.batchCollector.tupleEmitted || this.batchCollector.batchEmitted) { break; } } }
SpoutOutputBatcher implements IRichSpout { @Override public void nextTuple() { while(true) { this.batchCollector.tupleEmitted = false; this.batchCollector.batchEmitted = false; this.wrappedSpout.nextTuple(); if(!this.batchCollector.tupleEmitted || this.batchCollector.batchEmitted) { break; } } } }
SpoutOutputBatcher implements IRichSpout { @Override public void nextTuple() { while(true) { this.batchCollector.tupleEmitted = false; this.batchCollector.batchEmitted = false; this.wrappedSpout.nextTuple(); if(!this.batchCollector.tupleEmitted || this.batchCollector.batchEmitted) { break; } } } SpoutOutputBatcher(IRichSpout spout, int batchSize);  SpoutOutputBatcher(IRichSpout spout, Map<String, Integer> batchSizes); }
SpoutOutputBatcher implements IRichSpout { @Override public void nextTuple() { while(true) { this.batchCollector.tupleEmitted = false; this.batchCollector.batchEmitted = false; this.wrappedSpout.nextTuple(); if(!this.batchCollector.tupleEmitted || this.batchCollector.batchEmitted) { break; } } } SpoutOutputBatcher(IRichSpout spout, int batchSize);  SpoutOutputBatcher(IRichSpout spout, Map<String, Integer> batchSizes); @Override void open(@SuppressWarnings("rawtypes") Map conf, TopologyContext context, SpoutOutputCollector collector); @Override void close(); @Override void activate(); @Override void deactivate(); @Override void nextTuple(); @Override void ack(Object msgId); @Override void fail(Object msgId); @Override void declareOutputFields(OutputFieldsDeclarer declarer); @Override Map<String, Object> getComponentConfiguration(); static void registerKryoClasses(Config stormConfig); }
SpoutOutputBatcher implements IRichSpout { @Override public void nextTuple() { while(true) { this.batchCollector.tupleEmitted = false; this.batchCollector.batchEmitted = false; this.wrappedSpout.nextTuple(); if(!this.batchCollector.tupleEmitted || this.batchCollector.batchEmitted) { break; } } } SpoutOutputBatcher(IRichSpout spout, int batchSize);  SpoutOutputBatcher(IRichSpout spout, Map<String, Integer> batchSizes); @Override void open(@SuppressWarnings("rawtypes") Map conf, TopologyContext context, SpoutOutputCollector collector); @Override void close(); @Override void activate(); @Override void deactivate(); @Override void nextTuple(); @Override void ack(Object msgId); @Override void fail(Object msgId); @Override void declareOutputFields(OutputFieldsDeclarer declarer); @Override Map<String, Object> getComponentConfiguration(); static void registerKryoClasses(Config stormConfig);  }