Add support for SEDA style with Spring Integration message channels

Resolves #4719

Author: fmbenhassine

spring-batch-integration/src/main/java/org/springframework/batch/integration/support/MessageChannelItemReader.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2025-present 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.integration.support;
+
+import org.jspecify.annotations.Nullable;
+
+import org.springframework.batch.infrastructure.item.ItemReader;
+import org.springframework.integration.core.MessagingTemplate;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.util.Assert;
+
+/**
+ * An {@link ItemReader} implementation that receives items as messages from a message
+ * channel using a {@link MessagingTemplate}. This item reader enables SEDA (Staged
+ * Event-Driven Architecture) patterns in Spring Batch jobs by decoupling item production
+ * from item consumption through messaging.
+ *
+ * @param <T> the type of items to be read
+ * @author Mahmoud Ben Hassine
+ * @since 6.0.0
+ */
+public class MessageChannelItemReader<T> implements ItemReader<T> {
+
+	private final MessagingTemplate messagingTemplate;
+
+	private final Class<T> targetType;
+
+	private MessageChannel messageChannel;
+
+	/**
+	 * Create a new {@link MessageChannelItemReader} instance. Messages will be read from
+	 * the default destination of the provided {@link MessagingTemplate} which must not be
+	 * null.
+	 * @param messagingTemplate the messaging template to use for reading messages
+	 * @param targetType the target type of items to convert messages to
+	 */
+	public MessageChannelItemReader(MessagingTemplate messagingTemplate, Class<T> targetType) {
+		this.targetType = targetType;
+		this.messagingTemplate = messagingTemplate;
+		MessageChannel defaultDestination = messagingTemplate.getDefaultDestination();
+		Assert.notNull(defaultDestination, "MessagingTemplate must have a default destination configured");
+		this.messageChannel = defaultDestination;
+	}
+
+	/**
+	 * Create a new {@link MessageChannelItemReader} instance. Messages will be read from
+	 * the provided target message channel.
+	 * @param messagingTemplate the messaging template to use for receiving messages
+	 * @param messageChannel the message channel to read messages from
+	 * @param targetType the target type of items to convert messages to
+	 */
+	public MessageChannelItemReader(MessagingTemplate messagingTemplate, MessageChannel messageChannel,
+			Class<T> targetType) {
+		this.messagingTemplate = messagingTemplate;
+		this.messageChannel = messageChannel;
+		this.targetType = targetType;
+	}
+
+	/**
+	 * Set the source message channel.
+	 * @param messageChannel the message channel to read messages from
+	 */
+	public void setMessageChannel(MessageChannel messageChannel) {
+		this.messageChannel = messageChannel;
+	}
+
+	@Override
+	public @Nullable T read() throws Exception {
+		return this.messagingTemplate.receiveAndConvert(this.messageChannel, this.targetType);
+	}
+
+}

spring-batch-integration/src/main/java/org/springframework/batch/integration/support/MessageChannelItemWriter.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2025-present 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.integration.support;
+
+import java.util.function.Consumer;
+
+import org.springframework.batch.infrastructure.item.Chunk;
+import org.springframework.batch.infrastructure.item.ItemWriter;
+import org.springframework.integration.core.MessagingTemplate;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.util.Assert;
+
+/**
+ * An {@link ItemWriter} implementation that sends items as messages to a message channel
+ * using a {@link MessagingTemplate}. This item writer enables SEDA (Staged Event-Driven
+ * Architecture) patterns in Spring Batch jobs by decoupling item production from item
+ * consumption through messaging.
+ *
+ * @param <T> the type of items to be written
+ * @author Mahmoud Ben Hassine
+ * @since 6.0.0
+ */
+public class MessageChannelItemWriter<T> implements ItemWriter<T> {
+
+	private final MessagingTemplate messagingTemplate;
+
+	private MessageChannel messageChannel;
+
+	/**
+	 * Create a new {@link MessageChannelItemWriter} instance. Messages will be sent to
+	 * the default destination of the provided {@link MessagingTemplate} which must not be
+	 * null.
+	 * @param messagingGateway the messaging template to use for sending messages
+	 */
+	public MessageChannelItemWriter(MessagingTemplate messagingGateway) {
+		this.messagingTemplate = messagingGateway;
+		MessageChannel defaultDestination = messagingGateway.getDefaultDestination();
+		Assert.notNull(defaultDestination, "MessagingTemplate must have a default destination configured");
+		this.messageChannel = defaultDestination;
+	}
+
+	/**
+	 * Create a new {@link MessageChannelItemWriter} instance. Messages will be sent to
+	 * the provided message channel.
+	 * @param messagingTemplate the messaging template to use for sending messages
+	 * @param messageChannel the message channel to send messages to
+	 */
+	public MessageChannelItemWriter(MessagingTemplate messagingTemplate, MessageChannel messageChannel) {
+		this.messagingTemplate = messagingTemplate;
+		this.messageChannel = messageChannel;
+	}
+
+	/**
+	 * Set the target message channel.
+	 * @param messageChannel the message channel to send messages to
+	 */
+	public void setMessageChannel(MessageChannel messageChannel) {
+		this.messageChannel = messageChannel;
+	}
+
+	@Override
+	public void write(Chunk<? extends T> items) throws Exception {
+		if (!items.isEmpty()) {
+			items.forEach((Consumer<T>) t -> this.messagingTemplate.send(this.messageChannel, new GenericMessage<>(t)));
+		}
+	}
+
+}

spring-batch-integration/src/test/java/org/springframework/batch/integration/support/MessageChannelItemReaderTests.java

@@ -0,0 +1,31 @@
+package org.springframework.batch.integration.support;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import org.springframework.integration.core.MessagingTemplate;
+import org.springframework.messaging.MessageChannel;
+
+import static org.mockito.Mockito.mock;
+
+class MessageChannelItemReaderTests {
+
+	@Test
+	void testRead() throws Exception {
+		// given
+		MessageChannel messageChannel = mock();
+		MessagingTemplate messagingTemplate = mock();
+		Mockito.when(messagingTemplate.receiveAndConvert(messageChannel, String.class)).thenReturn("Foo");
+		MessageChannelItemReader<String> reader = new MessageChannelItemReader<>(messagingTemplate, messageChannel,
+				String.class);
+
+		// when
+		String item = reader.read();
+
+		// then
+		Mockito.verify(messagingTemplate, Mockito.times(1)).receiveAndConvert(messageChannel, String.class);
+		Assertions.assertEquals("Foo", item);
+	}
+
+}

spring-batch-integration/src/test/java/org/springframework/batch/integration/support/MessageChannelItemWriterTests.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2025-present 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.integration.support;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import org.springframework.batch.infrastructure.item.Chunk;
+import org.springframework.integration.core.MessagingTemplate;
+import org.springframework.messaging.MessageChannel;
+
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Tests for {@link MessageChannelItemWriter}.
+ *
+ * @author Mahmoud Ben Hassine
+ */
+class MessageChannelItemWriterTests {
+
+	@Test
+	void testWrite() throws Exception {
+		// given
+		MessageChannel messageChannel = mock();
+		MessagingTemplate messagingTemplate = mock();
+		MessageChannelItemWriter<String> itemWriter = new MessageChannelItemWriter<>(messagingTemplate, messageChannel);
+		Chunk<String> items = Chunk.of("foo", "bar");
+
+		// when
+		itemWriter.write(items);
+
+		// then
+		Mockito.verify(messagingTemplate, Mockito.times(2)).send(eq(messageChannel), Mockito.any());
+	}
+
+}