From c765e80cd3cf24b20e80df2422f950211ea8e12f Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@vmware.com>
Date: Tue, 7 Jul 2020 14:35:55 -0400
Subject: [PATCH] SO-62761903: Inject BF into Gateway's correlator

Related to https://stackoverflow.com/questions/62761903/spring-integration-reactive-streams-support-exception-in-creating-a-reactive

The `MessagingGatewaySupport` creates an internal endpoint
for consuming messages from the provided `replyChannel`.
The endpoint type depends on the channel type.
The `ReactiveStreamsConsumer` was missing a `BeanFactory` injection
causing an error when `ReactiveStreamsConsumer` tries to extract a default `ErrorHandler`
from `BeanFactory`

* Refactor `MessagingGatewaySupport` to inject a `BeanFactory` to all
the possible correlator endpoints.
Also always call `afterPropertiesSet()` for all of them

**Cherry-pick to 5.3.x & 5.2.x**
---
 .../gateway/MessagingGatewaySupport.java       | 14 ++++++--------
 .../gateway/GatewayProxyFactoryBeanTests.java  | 18 ++++++++++++++++++
 2 files changed, 24 insertions(+), 8 deletions(-)

diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java
index 20b0161984..437037423b 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java
@@ -787,23 +787,21 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 				}
 				else if (replyChan instanceof PollableChannel) {
 					PollingConsumer endpoint = new PollingConsumer((PollableChannel) replyChan, handler);
-					if (beanFactory != null) {
-						endpoint.setBeanFactory(beanFactory);
-					}
 					endpoint.setReceiveTimeout(this.replyTimeout);
-					endpoint.afterPropertiesSet();
 					correlator = endpoint;
 				}
 				else if (replyChan instanceof ReactiveStreamsSubscribableChannel) {
-					ReactiveStreamsConsumer endpoint =
-							new ReactiveStreamsConsumer(replyChan, (Subscriber<Message<?>>) handler);
-					endpoint.afterPropertiesSet();
-					correlator = endpoint;
+					correlator = new ReactiveStreamsConsumer(replyChan, (Subscriber<Message<?>>) handler);
 				}
 				else {
 					throw new MessagingException("Unsupported 'replyChannel' type [" + replyChan.getClass() + "]."
 							+ "SubscribableChannel or PollableChannel type are supported.");
 				}
+
+				if (beanFactory != null) {
+					correlator.setBeanFactory(beanFactory);
+				}
+				correlator.afterPropertiesSet();
 				this.replyMessageCorrelator = correlator;
 			}
 			if (isRunning()) {
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java
index e21ce82121..3c39964064 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java
@@ -46,6 +46,7 @@ import org.springframework.expression.common.LiteralExpression;
 import org.springframework.integration.annotation.Gateway;
 import org.springframework.integration.annotation.GatewayHeader;
 import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.FluxMessageChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
 import org.springframework.integration.support.utils.IntegrationUtils;
@@ -158,6 +159,23 @@ public class GatewayProxyFactoryBeanTests {
 		assertThat(message.getPayload()).isEqualTo("foo");
 	}
 
+	@Test
+	public void testReactiveReplyChannel() {
+		QueueChannel requestChannel = new QueueChannel();
+		startResponder(requestChannel);
+		FluxMessageChannel replyChannel = new FluxMessageChannel();
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestService.class);
+		proxyFactory.setDefaultRequestChannel(requestChannel);
+		proxyFactory.setDefaultReplyChannel(replyChannel);
+
+		proxyFactory.setBeanFactory(mock(BeanFactory.class));
+		proxyFactory.afterPropertiesSet();
+		TestService service = (TestService) proxyFactory.getObject();
+
+		String result = service.requestReply("test");
+		assertThat(result).isEqualTo("testbar");
+	}
+
 	@Test
 	public void testRequestReplyWithTypeConversion() {
 		final QueueChannel requestChannel = new QueueChannel();