Introduce Reactive Message producer

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java

@@ -16,11 +16,14 @@
 
 package org.springframework.integration.endpoint;
 
+import org.reactivestreams.Publisher;
+
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.SmartInitializingSingleton;
 import org.springframework.core.AttributeAccessor;
 import org.springframework.integration.IntegrationPattern;
 import org.springframework.integration.IntegrationPatternType;
+import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.history.MessageHistory;
@@ -36,6 +39,10 @@
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
+import reactor.core.Disposable;
+import reactor.core.Disposables;
+import reactor.core.publisher.Flux;
+
 /**
  * A support class for producer endpoints that provides a setter for the
  * output channel and a convenience method for sending Messages.
@@ -47,6 +54,8 @@
 public abstract class MessageProducerSupport extends AbstractEndpoint implements MessageProducer, TrackableComponent,
 		SmartInitializingSingleton, IntegrationPattern {
 
+	private final Disposable.Composite reactiveSubscriptions = Disposables.composite();
+
 	private final MessagingTemplate messagingTemplate = new MessagingTemplate();
 
 	private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy();
@@ -196,13 +205,10 @@ protected void sendMessage(Message<?> messageArg) {
 		if (message == null) {
 			throw new MessagingException("cannot send a null message");
 		}
-		if (this.shouldTrack) {
-			message = MessageHistory.write(message, this, getMessageBuilderFactory());
-		}
+		message = trackMessageIfAny(message);
 		try {
-			MessageChannel messageChannel = getOutputChannel();
-			Assert.state(messageChannel != null, "The 'outputChannel' or `outputChannelName` must be configured");
-			this.messagingTemplate.send(messageChannel, message);
+			MessageChannel outputChannel = getRequiredOutputChannel();
+			this.messagingTemplate.send(outputChannel, message);
 		}
 		catch (RuntimeException ex) {
 			if (!sendErrorMessageIfNecessary(message, ex)) {
@@ -211,14 +217,37 @@ protected void sendMessage(Message<?> messageArg) {
 		}
 	}
 
+	protected void subscribeToPublisher(Publisher<? extends Message<?>> publisher) {
+		MessageChannel outputChannel = getRequiredOutputChannel();
+		Flux<Message<?>> messageFlux =
+				Flux.from(publisher)
+						.map(this::trackMessageIfAny);
+		if (outputChannel instanceof ReactiveStreamsSubscribableChannel) {
+			((ReactiveStreamsSubscribableChannel) outputChannel).subscribeTo(messageFlux);
+		}
+		else {
+			this.reactiveSubscriptions.add(
+					messageFlux
+							.doOnNext((message) -> {
+								try {
+									sendMessage(message);
+								}
+								catch (Exception ex) {
+									logger.error("Error sending a message: " + message, ex);
+								}
+							})
+							.subscribe());
+		}
+	}
+
 	/**
 	 * Send an error message based on the exception and message.
 	 * @param message the message.
 	 * @param exception the exception.
 	 * @return true if the error channel is available and message sent.
 	 * @since 4.3.10
 	 */
-	protected final boolean sendErrorMessageIfNecessary(Message<?> message, RuntimeException exception) {
+	protected final boolean sendErrorMessageIfNecessary(Message<?> message, Exception exception) {
 		MessageChannel channel = getErrorChannel();
 		if (channel != null) {
 			this.messagingTemplate.send(channel, buildErrorMessage(message, exception));
@@ -235,9 +264,8 @@ protected final boolean sendErrorMessageIfNecessary(Message<?> message, RuntimeE
 	 * @return the error message.
 	 * @since 4.3.10
 	 */
-	protected final ErrorMessage buildErrorMessage(Message<?> message, RuntimeException exception) {
-		return this.errorMessageStrategy.buildErrorMessage(exception,
-				getErrorMessageAttributes(message));
+	protected final ErrorMessage buildErrorMessage(Message<?> message, Exception exception) {
+		return this.errorMessageStrategy.buildErrorMessage(exception, getErrorMessageAttributes(message));
 	}
 
 	/**
@@ -252,4 +280,25 @@ protected AttributeAccessor getErrorMessageAttributes(Message<?> message) {
 		return ErrorMessageUtils.getAttributeAccessor(message, null);
 	}
 
+	private MessageChannel getRequiredOutputChannel() {
+		MessageChannel messageChannel = getOutputChannel();
+		Assert.state(messageChannel != null, "The 'outputChannel' or `outputChannelName` must be configured");
+		return messageChannel;
+	}
+
+	private Message<?> trackMessageIfAny(Message<?> message) {
+		if (this.shouldTrack) {
+			return MessageHistory.write(message, this, getMessageBuilderFactory());
+		}
+		else {
+			return message;
+		}
+	}
+
+	@Override
+	public void destroy() {
+		this.reactiveSubscriptions.dispose();
+		super.destroy();
+	}
+
 }

spring-integration-core/src/test/java/org/springframework/integration/endpoint/ReactiveMessageProducerTests.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2020 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.integration.endpoint;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.channel.FluxMessageChannel;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 5.3
+ */
+@SpringJUnitConfig
+@DirtiesContext
+public class ReactiveMessageProducerTests {
+
+	@Autowired
+	public FluxMessageChannel fluxMessageChannel;
+
+	@Test
+	public void test() {
+		StepVerifier.create(
+				Flux.from(this.fluxMessageChannel)
+						.map(Message::getPayload)
+						.cast(String.class))
+				.expectNext("test1", "test2")
+				.thenCancel()
+				.verify();
+	}
+
+	@Configuration
+	@EnableIntegration
+	public static class Config {
+
+		@Bean
+		public FluxMessageChannel fluxMessageChannel() {
+			return new FluxMessageChannel();
+		}
+
+		@Bean
+		public MessageProducerSupport producer() {
+			MessageProducerSupport producer =
+					new MessageProducerSupport() {
+
+						@Override
+						protected void doStart() {
+							subscribeToPublisher(Flux.just("test1", "test2").map(GenericMessage::new));
+						}
+
+					};
+			producer.setOutputChannel(fluxMessageChannel());
+			return producer;
+		}
+
+	}
+
+}