Add composite item reader implementation

Resolves #757

Author: fmbenhassine

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/CompositeItemReader.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.item.support;
+
+import java.util.Iterator;
+import java.util.List;
+
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.ItemStreamException;
+import org.springframework.batch.item.ItemStreamReader;
+
+/**
+ * Composite reader that delegates reading to a list of {@link ItemStreamReader}s. This
+ * implementation is not thread-safe.
+ *
+ * @author Mahmoud Ben Hassine
+ * @param <T> type of objects to read
+ * @since 5.2
+ */
+public class CompositeItemReader<T> implements ItemStreamReader<T> {
+
+	private final List<ItemStreamReader<T>> delegates;
+
+	private final Iterator<ItemStreamReader<T>> delegatesIterator;
+
+	private ItemStreamReader<T> currentDelegate;
+
+	/**
+	 * Create a new {@link CompositeItemReader}.
+	 * @param delegates the delegate readers to read data
+	 */
+	public CompositeItemReader(List<ItemStreamReader<T>> delegates) {
+		this.delegates = delegates;
+		this.delegatesIterator = this.delegates.iterator();
+		this.currentDelegate = this.delegatesIterator.hasNext() ? this.delegatesIterator.next() : null;
+	}
+
+	// TODO: check if we need to open/close delegates on the fly in read() to avoid
+	// opening resources early for a long time
+	@Override
+	public void open(ExecutionContext executionContext) throws ItemStreamException {
+		for (ItemStreamReader<T> delegate : delegates) {
+			delegate.open(executionContext);
+		}
+	}
+
+	@Override
+	public T read() throws Exception {
+		if (this.currentDelegate == null) {
+			return null;
+		}
+		T item = currentDelegate.read();
+		if (item == null) {
+			currentDelegate = this.delegatesIterator.hasNext() ? this.delegatesIterator.next() : null;
+			return read();
+		}
+		return item;
+	}
+
+	@Override
+	public void update(ExecutionContext executionContext) throws ItemStreamException {
+		if (this.currentDelegate != null) {
+			this.currentDelegate.update(executionContext);
+		}
+	}
+
+	@Override
+	public void close() throws ItemStreamException {
+		for (ItemStreamReader<T> delegate : delegates) {
+			delegate.close();
+		}
+	}
+
+}

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/CompositeItemReaderTests.java

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.item.support;
+
+import java.util.Arrays;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.ItemStreamReader;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+/**
+ * Test class for {@link CompositeItemReader}.
+ *
+ * @author Mahmoud Ben Hassine
+ */
+public class CompositeItemReaderTests {
+
+	@Test
+	void testCompositeItemReaderOpen() {
+		// given
+		ItemStreamReader<String> reader1 = mock();
+		ItemStreamReader<String> reader2 = mock();
+		CompositeItemReader<String> compositeItemReader = new CompositeItemReader<>(Arrays.asList(reader1, reader2));
+		ExecutionContext executionContext = new ExecutionContext();
+
+		// when
+		compositeItemReader.open(executionContext);
+
+		// then
+		verify(reader1).open(executionContext);
+		verify(reader2).open(executionContext);
+	}
+
+	@Test
+	void testCompositeItemReaderRead() throws Exception {
+		// given
+		ItemStreamReader<String> reader1 = mock();
+		ItemStreamReader<String> reader2 = mock();
+		CompositeItemReader<String> compositeItemReader = new CompositeItemReader<>(Arrays.asList(reader1, reader2));
+		when(reader1.read()).thenReturn("foo1", "foo2", null);
+		when(reader2.read()).thenReturn("bar1", "bar2", null);
+
+		// when & then
+		compositeItemReader.read();
+		verify(reader1, times(1)).read();
+		compositeItemReader.read();
+		verify(reader1, times(2)).read();
+		compositeItemReader.read();
+		verify(reader1, times(3)).read();
+
+		compositeItemReader.read();
+		verify(reader2, times(2)).read();
+		compositeItemReader.read();
+		verify(reader2, times(3)).read();
+		compositeItemReader.read();
+		verify(reader2, times(3)).read();
+	}
+
+	@Test
+	void testCompositeItemReaderUpdate() {
+		// given
+		ItemStreamReader<String> reader1 = mock();
+		ItemStreamReader<String> reader2 = mock();
+		CompositeItemReader<String> compositeItemReader = new CompositeItemReader<>(Arrays.asList(reader1, reader2));
+		ExecutionContext executionContext = new ExecutionContext();
+
+		// when
+		compositeItemReader.update(executionContext);
+
+		// then
+		verify(reader1).update(executionContext);
+		verifyNoInteractions(reader2); // reader1 is the current delegate in this setup
+	}
+
+	@Test
+	void testCompositeItemReaderClose() {
+		// given
+		ItemStreamReader<String> reader1 = mock();
+		ItemStreamReader<String> reader2 = mock();
+		CompositeItemReader<String> compositeItemReader = new CompositeItemReader<>(Arrays.asList(reader1, reader2));
+
+		// when
+		compositeItemReader.close();
+
+		// then
+		verify(reader1).close();
+		verify(reader2).close();
+	}
+
+}

spring-batch-samples/README.md

@@ -26,6 +26,7 @@ Here is a list of samples with checks to indicate which features each one demons
 | [Hello world Job Sample](#hello-world-job-sample)             |      |       |         |                   |               |            |            |              |                |       X        |           |
 | [Amqp Job Sample](#amqp-job-sample)                           |      |       |         |                   |               |            |            |              |                |       X        |           |
 | [BeanWrapperMapper Sample](#beanwrappermapper-sample)         |      |       |         |         X         |               |            |            |              |                |                |           |
+| [Composite ItemReader Sample](#composite-itemreader-sample)        |      |       |         |                   |               |            |     X      |              |                |                |           |
 | [Composite ItemWriter Sample](#composite-itemwriter-sample)   |      |       |         |                   |               |            |     X      |              |                |                |           |
 | [Customer Filter Sample](#customer-filter-sample)             |      |       |         |                   |               |            |            |              |                |                |     X     |
 | [Reader Writer Adapter Sample](#reader-writer-adapter-sample) |      |       |         |                   |               |            |     X      |              |                |                |           |
@@ -121,6 +122,16 @@ prototype according to the field names in the file.
 
 [BeanWrapperMapper Sample](src/main/java/org/springframework/batch/samples/beanwrapper/README.md)
 
+### Composite ItemReader Sample
+
+This sample shows how to use a composite item reader to read data with
+the same format from different data sources.
+
+In this sample, data items of type `Person` are read from two flat files
+and a relational database table.
+
+[Composite reader Sample](src/main/java/org/springframework/batch/samples/compositereader/README.md)
+
 ### Composite ItemWriter Sample
 
 This shows a common use case using a composite pattern, composing

spring-batch-samples/src/main/java/org/springframework/batch/samples/compositereader/README.md

@@ -0,0 +1,18 @@
+## Composite ItemReader Sample
+
+### About
+
+This sample shows how to use a composite item reader to read data with
+the same format from different data sources.
+
+In this sample, data items of type `Person` are read from two flat files
+and a relational database table.
+
+### Run the sample
+
+You can run the sample from the command line as following:
+
+```
+$>cd spring-batch-samples
+$>../mvnw -Dtest=CompositeItemReaderSampleFunctionalTests#testJobLaunch test
+```

spring-batch-samples/src/main/resources/org/springframework/batch/samples/compositereader/data/persons1.csv

@@ -0,0 +1,2 @@
+1,foo1
+2,foo2

spring-batch-samples/src/main/resources/org/springframework/batch/samples/compositereader/data/persons2.csv

@@ -0,0 +1,2 @@
+3,bar1
+4,bar2

spring-batch-samples/src/main/resources/org/springframework/batch/samples/compositereader/sql/data.sql

@@ -0,0 +1,2 @@
+insert into person_source values (5, 'baz1');
+insert into person_source values (6, 'baz2');

spring-batch-samples/src/main/resources/org/springframework/batch/samples/compositereader/sql/schema.sql

@@ -0,0 +1,2 @@
+create table person_source (id int primary key, name varchar(20));
+create table person_target (id int primary key, name varchar(20));