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**

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java

@@ -787,23 +787,21 @@ protected void registerReplyMessageCorrelatorIfNecessary() {
 				}
 				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()) {

spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java

@@ -46,6 +46,7 @@
 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 void testReceiveMessage() {
 		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();