Implement ItemStreamWriter in MappingItemWriter

spring-batch-docs/modules/ROOT/pages/readers-and-writers/item-reader-writer-implementations.adoc

@@ -7,11 +7,9 @@ discussed in the previous sections.
 [[decorators]]
 == Decorators
 
-In some cases, a user needs specialized behavior to be appended to a pre-existing
-`ItemReader`.   Spring Batch offers some out of the box decorators that can add
-additional behavior to to your `ItemReader` and `ItemWriter` implementations.
-
-Spring Batch includes the following decorators:
+In some cases, you might need specialized behavior to be appended to a pre-existing
+`ItemReader` or `ItemWriter` implementation.
+For this purpose, Spring Batch offers the following out-of-the-box decorators:
 
 * xref:readers-and-writers/item-reader-writer-implementations.adoc#synchronizedItemStreamReader[`SynchronizedItemStreamReader`]
 * xref:readers-and-writers/item-reader-writer-implementations.adoc#singleItemPeekableItemReader[`SingleItemPeekableItemReader`]
@@ -113,10 +111,12 @@ through the provided `Classifier`. Spring Batch provides a
 
 The `MappingItemWriter` adapts an `ItemWriter` accepting items of a given type to one accepting
 items of another type by applying a mapping function to each item before writing.
+Thread-safety is guaranteed as long as the downstream item writer is thread-safe, and state
+management is honored with a downstream `ItemStream` item writer.
 
 This item writer is most useful when used in combination with a `CompositeItemWriter`, where the
 mapping function in front of the downstream writer can be a getter of the input item or a more
-complex transformation logic.
+complex transformation logic, effectively allowing deconstruction patterns.
 
 [[messagingReadersAndWriters]]
 == Messaging Readers And Writers

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

@@ -17,6 +17,10 @@
 package org.springframework.batch.infrastructure.item.support;
 
 import org.springframework.batch.infrastructure.item.Chunk;
+import org.springframework.batch.infrastructure.item.ExecutionContext;
+import org.springframework.batch.infrastructure.item.ItemStream;
+import org.springframework.batch.infrastructure.item.ItemStreamException;
+import org.springframework.batch.infrastructure.item.ItemStreamWriter;
 import org.springframework.batch.infrastructure.item.ItemWriter;
 
 import java.util.function.Function;
@@ -25,10 +29,14 @@
  * Adapts an {@link ItemWriter} accepting items of type {@link U} to one accepting items
  * of type {@link T} by applying a mapping function to each item before writing.
  * <p>
- * The {@code MappingItemWriter} is most useful when used in combination with a
+ * Thread-safety is guaranteed as long as the downstream item writer is thread-safe, and
+ * state management is honored with a downstream {@link ItemStream} item writer.
+ * <p>
+ * This adapter is most useful when used in combination with a
  * {@link org.springframework.batch.infrastructure.item.support.CompositeItemWriter},
  * where the mapping function in front of the downstream writer can be a getter of the
- * input item or a more complex transformation logic.
+ * input item or a more complex transformation logic, effectively allowing deconstruction
+ * patterns.
  * <p>
  * This adapter mimics the behavior of
  * {@link java.util.stream.Collectors#mapping(Function, java.util.stream.Collector)}.
@@ -38,7 +46,7 @@
  * @author Stefano Cordio
  * @since 6.0
  */
-public class MappingItemWriter<T, U> implements ItemWriter<T> {
+public class MappingItemWriter<T, U> implements ItemStreamWriter<T> {
 
 	private final Function<? super T, ? extends U> mapper;
 
@@ -59,4 +67,25 @@ public void write(Chunk<? extends T> chunk) throws Exception {
 		downstream.write(new Chunk<>(chunk.getItems().stream().map(mapper).toList()));
 	}
 
+	@Override
+	public void open(ExecutionContext executionContext) throws ItemStreamException {
+		if (downstream instanceof ItemStream itemStream) {
+			itemStream.open(executionContext);
+		}
+	}
+
+	@Override
+	public void update(ExecutionContext executionContext) throws ItemStreamException {
+		if (downstream instanceof ItemStream itemStream) {
+			itemStream.update(executionContext);
+		}
+	}
+
+	@Override
+	public void close() throws ItemStreamException {
+		if (downstream instanceof ItemStream itemStream) {
+			itemStream.close();
+		}
+	}
+
 }

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

@@ -17,12 +17,18 @@
 package org.springframework.batch.infrastructure.item.support;
 
 import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
 import org.springframework.batch.infrastructure.item.Chunk;
+import org.springframework.batch.infrastructure.item.ExecutionContext;
+import org.springframework.batch.infrastructure.item.ItemStreamWriter;
+import org.springframework.batch.infrastructure.item.ItemWriter;
 
 import java.util.function.Function;
 
+import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
 
 /**
  * @author Stefano Cordio
@@ -33,7 +39,7 @@ class MappingItemWriterTests {
 	void testWithMapperAcceptingItemSuperclass() throws Exception {
 		// given
 		Function<Entity, String> mapper = Entity::name;
-		ListItemWriter<String> downstream = mock();
+		ItemWriter<String> downstream = mock();
 		MappingItemWriter<Person, String> underTest = new MappingItemWriter<>(mapper, downstream);
 
 		// when
@@ -47,7 +53,7 @@ void testWithMapperAcceptingItemSuperclass() throws Exception {
 	void testWithMapperProducingMappedItemSubclass() throws Exception {
 		// given
 		Function<Person, String> mapper = Person::name;
-		ListItemWriter<CharSequence> downstream = mock();
+		ItemWriter<CharSequence> downstream = mock();
 		MappingItemWriter<Person, CharSequence> underTest = new MappingItemWriter<>(mapper, downstream);
 
 		// when
@@ -61,7 +67,7 @@ void testWithMapperProducingMappedItemSubclass() throws Exception {
 	void testWithDownstreamAcceptingMappedItemSuperclass() throws Exception {
 		// given
 		Function<Person, String> mapper = Person::name;
-		ListItemWriter<CharSequence> downstream = mock();
+		ItemWriter<CharSequence> downstream = mock();
 		MappingItemWriter<Person, String> underTest = new MappingItemWriter<>(mapper, downstream);
 
 		// when
@@ -71,6 +77,41 @@ void testWithDownstreamAcceptingMappedItemSuperclass() throws Exception {
 		verify(downstream).write(Chunk.of("Foo"));
 	}
 
+	@Test
+	void testWithDownstreamNotImplementingItemStream() {
+		// given
+		ExecutionContext executionContext = mock();
+		ItemWriter<Object> downstream = mock();
+		MappingItemWriter<Object, Object> underTest = new MappingItemWriter<>(Function.identity(), downstream);
+
+		// when
+		underTest.open(executionContext);
+		underTest.update(executionContext);
+		underTest.close();
+
+		// then
+		verifyNoInteractions(downstream);
+	}
+
+	@Test
+	void testWithDownstreamImplementingItemStream() {
+		// given
+		ExecutionContext executionContext = mock();
+		ItemStreamWriter<Object> downstream = mock();
+		MappingItemWriter<Object, Object> underTest = new MappingItemWriter<>(Function.identity(), downstream);
+
+		// when
+		underTest.open(executionContext);
+		underTest.update(executionContext);
+		underTest.close();
+
+		// then
+		InOrder inOrder = inOrder(downstream);
+		inOrder.verify(downstream).open(executionContext);
+		inOrder.verify(downstream).update(executionContext);
+		inOrder.verify(downstream).close();
+	}
+
 	private interface Entity {
 
 		String name();