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>) handler); - endpoint.afterPropertiesSet(); - correlator = endpoint; + correlator = new ReactiveStreamsConsumer(replyChan, (Subscriber>) 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();