TridentFileReadSpout implements IBatchSpout { @Override public void emitBatch(long batchId, TridentCollector collector) { List<String> batch = batches.get(batchId); if (batch == null) { batch = new ArrayList<String>(); for (int i = 0; i < maxBatchSize; i++) { batch.add(reader.nextLine()); } batches.put(batchId, batch); } for(String line : batch){ collector.emit(new Values(line)); } } TridentFileReadSpout(int maxBatchSize);  TridentFileReadSpout(int maxBatchSize, String file);  TridentFileReadSpout(int maxBatchSize, FileReader reader); @Override void open(Map conf, TopologyContext context); @Override void emitBatch(long batchId, TridentCollector collector); @Override void ack(long batchId); @Override void close(); @Override Map getComponentConfiguration(); @Override Fields getOutputFields(); static final String FIELDS; }
@Test public void testEmitBatch() { spout.emitBatch(0L, collector); spout.emitBatch(1L, collector); verify(collector, times(20)).emit(any(Values.class)); Map<Long, List<String>> batches = spout.getBatches(); assertThat(batches) .hasSize(2) .containsKey(0L) .containsKey(1L); assertThat(batches.get(0L)).hasSize(10); assertThat(batches.get(1L)).hasSize(10); }