@SuppressWarnings({"unchecked", "rawtypes"}) @Test public void batchEmitDirect() throws IllegalArgumentException, IllegalAccessException { TopologyContext context = mock(TopologyContext.class); ISpoutOutputCollector col = mock(ISpoutOutputCollector.class); BatchSpoutOutputCollector collector = new BatchSpoutOutputCollector(context, col, 0); SpoutBatchCollectorImpl collectorImpl = new SpoutBatchCollectorImpl(collector, context, 0); int taskId = 0; String streamId = new String(); Batch batch = mock(Batch.class); Object messageId = mock(Object.class); collector.batchEmitted = false; collectorImpl.doEmitDirect(taskId, streamId, null, batch, messageId); verify(col).emitDirect(taskId, streamId, (List)batch, messageId); Assert.assertTrue(collector.batchEmitted); }
@SuppressWarnings({"unchecked", "rawtypes"}) @Override protected void doEmitDirect(int taskId, String streamId, Collection<Tuple> anchors, Object tupleOrBatch, Object messageId) { assert (anchors == null); logger.trace("taskId: {}; streamId: {}; tuple/batch: {}; messageId: {}", new Integer(taskId), streamId, tupleOrBatch, messageId); this.spoutBatchCollector.batchEmitted = true; this.spoutBatchCollector.collector.emitDirect(taskId, streamId, (List)tupleOrBatch, messageId); }
SpoutBatchCollectorImpl extends AbstractBatchCollector { @SuppressWarnings({"unchecked", "rawtypes"}) @Override protected void doEmitDirect(int taskId, String streamId, Collection<Tuple> anchors, Object tupleOrBatch, Object messageId) { assert (anchors == null); logger.trace("taskId: {}; streamId: {}; tuple/batch: {}; messageId: {}", new Integer(taskId), streamId, tupleOrBatch, messageId); this.spoutBatchCollector.batchEmitted = true; this.spoutBatchCollector.collector.emitDirect(taskId, streamId, (List)tupleOrBatch, messageId); } }
SpoutBatchCollectorImpl extends AbstractBatchCollector { @SuppressWarnings({"unchecked", "rawtypes"}) @Override protected void doEmitDirect(int taskId, String streamId, Collection<Tuple> anchors, Object tupleOrBatch, Object messageId) { assert (anchors == null); logger.trace("taskId: {}; streamId: {}; tuple/batch: {}; messageId: {}", new Integer(taskId), streamId, tupleOrBatch, messageId); this.spoutBatchCollector.batchEmitted = true; this.spoutBatchCollector.collector.emitDirect(taskId, streamId, (List)tupleOrBatch, messageId); } SpoutBatchCollectorImpl(BatchSpoutOutputCollector spoutBatchCollector, TopologyContext context, int batchSize);  SpoutBatchCollectorImpl(BatchSpoutOutputCollector spoutBatchCollector, TopologyContext context,
		Map<String, Integer> batchSizes); }
SpoutBatchCollectorImpl extends AbstractBatchCollector { @SuppressWarnings({"unchecked", "rawtypes"}) @Override protected void doEmitDirect(int taskId, String streamId, Collection<Tuple> anchors, Object tupleOrBatch, Object messageId) { assert (anchors == null); logger.trace("taskId: {}; streamId: {}; tuple/batch: {}; messageId: {}", new Integer(taskId), streamId, tupleOrBatch, messageId); this.spoutBatchCollector.batchEmitted = true; this.spoutBatchCollector.collector.emitDirect(taskId, streamId, (List)tupleOrBatch, messageId); } SpoutBatchCollectorImpl(BatchSpoutOutputCollector spoutBatchCollector, TopologyContext context, int batchSize);  SpoutBatchCollectorImpl(BatchSpoutOutputCollector spoutBatchCollector, TopologyContext context,
		Map<String, Integer> batchSizes);  }
SpoutBatchCollectorImpl extends AbstractBatchCollector { @SuppressWarnings({"unchecked", "rawtypes"}) @Override protected void doEmitDirect(int taskId, String streamId, Collection<Tuple> anchors, Object tupleOrBatch, Object messageId) { assert (anchors == null); logger.trace("taskId: {}; streamId: {}; tuple/batch: {}; messageId: {}", new Integer(taskId), streamId, tupleOrBatch, messageId); this.spoutBatchCollector.batchEmitted = true; this.spoutBatchCollector.collector.emitDirect(taskId, streamId, (List)tupleOrBatch, messageId); } SpoutBatchCollectorImpl(BatchSpoutOutputCollector spoutBatchCollector, TopologyContext context, int batchSize);  SpoutBatchCollectorImpl(BatchSpoutOutputCollector spoutBatchCollector, TopologyContext context,
		Map<String, Integer> batchSizes);   }