CompositeRecordProcessor implements RecordProcessor<I, O> { @Override @SuppressWarnings(value = "unchecked") public Record<O> processRecord(Record<I> record) throws Exception { Record processedRecord = record; for (RecordProcessor processor : processors) { processedRecord = processor.processRecord(processedRecord); if (processedRecord == null) { return null; } } return processedRecord; } CompositeRecordProcessor();  CompositeRecordProcessor(List<RecordProcessor<I, O>> processors); @Override @SuppressWarnings(value = "unchecked") Record<O> processRecord(Record<I> record); void addRecordProcessor(RecordProcessor<I, O> recordProcessor);  }
@Test public void recordsShouldBeProcessedByDelegates() throws Exception { when(recordProcessor1.processRecord(record)).thenReturn(processedRecord); when(recordProcessor2.processRecord(processedRecord)).thenReturn(processedRecord); Record<String> actual = compositeRecordProcessor.processRecord(record); assertThat(actual).isEqualTo(processedRecord); }
@Test public void whenProcessorReturnsNull_thenNextProcessorsShouldBeSkipped() throws Exception { when(recordProcessor1.processRecord(record)).thenReturn(null); Record<String> actual = compositeRecordProcessor.processRecord(record); assertThat(actual).isNull(); }