@SuppressWarnings({"unchecked", "rawtypes"}) @Test public void testBatchEmit() throws IllegalArgumentException, IllegalAccessException { IOutputCollector col = mock(IOutputCollector.class); BatchOutputCollector collector = mock(BatchOutputCollector.class); PowerMockito.field(BatchOutputCollector.class, "collector").set(collector, col); BoltBatchCollectorImpl collectorImpl = new BoltBatchCollectorImpl(collector, mock(TopologyContext.class), 0); String streamId = new String(); Collection<Tuple> anchors = mock(Collection.class); Batch batch = mock(Batch.class); collectorImpl.doEmit(streamId, anchors, batch, null); verify(col).emit(streamId, anchors, (List)batch); }
@SuppressWarnings({"rawtypes", "unchecked"}) @Override protected List<Integer> doEmit(String streamId, Collection<Tuple> anchors, Object tupleOrBatch, Object messageId) { assert (messageId == null); BatchOutputCollector.logger .trace("streamId: {}; anchors: {}, tuple/batch: {}", streamId, anchors, tupleOrBatch); return this.boltBatchCollector.collector.emit(streamId, anchors, (List)tupleOrBatch); }
BoltBatchCollectorImpl extends AbstractBatchCollector { @SuppressWarnings({"rawtypes", "unchecked"}) @Override protected List<Integer> doEmit(String streamId, Collection<Tuple> anchors, Object tupleOrBatch, Object messageId) { assert (messageId == null); BatchOutputCollector.logger .trace("streamId: {}; anchors: {}, tuple/batch: {}", streamId, anchors, tupleOrBatch); return this.boltBatchCollector.collector.emit(streamId, anchors, (List)tupleOrBatch); } }
BoltBatchCollectorImpl extends AbstractBatchCollector { @SuppressWarnings({"rawtypes", "unchecked"}) @Override protected List<Integer> doEmit(String streamId, Collection<Tuple> anchors, Object tupleOrBatch, Object messageId) { assert (messageId == null); BatchOutputCollector.logger .trace("streamId: {}; anchors: {}, tuple/batch: {}", streamId, anchors, tupleOrBatch); return this.boltBatchCollector.collector.emit(streamId, anchors, (List)tupleOrBatch); } BoltBatchCollectorImpl(BatchOutputCollector boltBatchCollector, TopologyContext context, int batchSize);  BoltBatchCollectorImpl(BatchOutputCollector boltBatchCollector, TopologyContext context,
		Map<String, Integer> batchSizes); }
BoltBatchCollectorImpl extends AbstractBatchCollector { @SuppressWarnings({"rawtypes", "unchecked"}) @Override protected List<Integer> doEmit(String streamId, Collection<Tuple> anchors, Object tupleOrBatch, Object messageId) { assert (messageId == null); BatchOutputCollector.logger .trace("streamId: {}; anchors: {}, tuple/batch: {}", streamId, anchors, tupleOrBatch); return this.boltBatchCollector.collector.emit(streamId, anchors, (List)tupleOrBatch); } BoltBatchCollectorImpl(BatchOutputCollector boltBatchCollector, TopologyContext context, int batchSize);  BoltBatchCollectorImpl(BatchOutputCollector boltBatchCollector, TopologyContext context,
		Map<String, Integer> batchSizes);  }
BoltBatchCollectorImpl extends AbstractBatchCollector { @SuppressWarnings({"rawtypes", "unchecked"}) @Override protected List<Integer> doEmit(String streamId, Collection<Tuple> anchors, Object tupleOrBatch, Object messageId) { assert (messageId == null); BatchOutputCollector.logger .trace("streamId: {}; anchors: {}, tuple/batch: {}", streamId, anchors, tupleOrBatch); return this.boltBatchCollector.collector.emit(streamId, anchors, (List)tupleOrBatch); } BoltBatchCollectorImpl(BatchOutputCollector boltBatchCollector, TopologyContext context, int batchSize);  BoltBatchCollectorImpl(BatchOutputCollector boltBatchCollector, TopologyContext context,
		Map<String, Integer> batchSizes);   }