Refinement for gateway Mono processing (#3075)

* Refinement for gateway Mono processing

* Introduce a `MessagingGatewaySupport.MonoReplyChannel` instead of
`FutureReplyChannel` for better on demand handling and reusing a
`Mono` returned from the target handler
* Refactor `GatewayProxyFactoryBean` to identify a target return type
(including generics for `Function`) during initialization
* Handle a `Mono` return type via
`MessagingGatewaySupport.sendAndReceiveMessageReactive()`
* Some `@Nullable` in the `GatewayProxyFactoryBean` and `ExpressionUtils`
* Add `MonoFunction` test-case into the `FunctionsTests.kt`

* * Deprecate `GatewayProxyFactoryBean.setServiceInterface()` in favor of
ctor initialization
* Fix `GatewayProxyFactoryBean.setServiceInterface()` usage in tests

spring-integration-core/src/main/java/org/springframework/integration/expression/ExpressionUtils.java

@@ -46,6 +46,7 @@ import org.springframework.util.Assert;
  * @author Gary Russell
  * @author Oleg Zhurakousky
  * @author Artem Bilan
+ *
  * @since 2.2
  */
 public final class ExpressionUtils {
@@ -81,7 +82,7 @@ public final class ExpressionUtils {
 	 * @param beanFactory The bean factory.
 	 * @return The evaluation context.
 	 */
-	public static StandardEvaluationContext createStandardEvaluationContext(BeanFactory beanFactory) {
+	public static StandardEvaluationContext createStandardEvaluationContext(@Nullable BeanFactory beanFactory) {
 		if (beanFactory == null) {
 			logger.warn("Creating EvaluationContext with no beanFactory", new RuntimeException("No beanFactory"));
 		}
@@ -95,14 +96,14 @@ public final class ExpressionUtils {
 	 * @return The evaluation context.
 	 * @since 4.3.15
 	 */
-	public static SimpleEvaluationContext createSimpleEvaluationContext(BeanFactory beanFactory) {
+	public static SimpleEvaluationContext createSimpleEvaluationContext(@Nullable BeanFactory beanFactory) {
 		if (beanFactory == null) {
 			logger.warn("Creating EvaluationContext with no beanFactory", new RuntimeException("No beanFactory"));
 		}
 		return (SimpleEvaluationContext) doCreateContext(beanFactory, true);
 	}
 
-	private static EvaluationContext doCreateContext(BeanFactory beanFactory, boolean simple) {
+	private static EvaluationContext doCreateContext(@Nullable BeanFactory beanFactory, boolean simple) {
 		ConversionService conversionService = null;
 		EvaluationContext evaluationContext = null;
 		if (beanFactory != null) {
@@ -129,8 +130,8 @@ public final class ExpressionUtils {
 	 * @param simple true if simple.
 	 * @return the evaluation context.
 	 */
-	private static EvaluationContext createEvaluationContext(ConversionService conversionService,
-			BeanFactory beanFactory, boolean simple) {
+	private static EvaluationContext createEvaluationContext(@Nullable ConversionService conversionService,
+			@Nullable BeanFactory beanFactory, boolean simple) {
 
 		if (simple) {
 			Builder ecBuilder = SimpleEvaluationContext.forPropertyAccessors(
@@ -166,6 +167,7 @@ public final class ExpressionUtils {
 	 */
 	public static File expressionToFile(Expression expression, EvaluationContext evaluationContext,
 			@Nullable Message<?> message, String name) {
+
 		File file;
 		Object value = message == null
 				? expression.getValue(evaluationContext)

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

@@ -41,7 +41,6 @@ public class GatewayMessageHandler extends AbstractReplyProducingMessageHandler
 
 	public GatewayMessageHandler() {
 		this.gatewayProxyFactoryBean = new GatewayProxyFactoryBean();
-		this.gatewayProxyFactoryBean.setServiceInterface(RequestReplyExchanger.class);
 	}
 
 	public void setRequestChannel(MessageChannel requestChannel) {

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

@@ -17,8 +17,6 @@
 package org.springframework.integration.gateway;
 
 import java.lang.reflect.Method;
-import java.lang.reflect.ParameterizedType;
-import java.lang.reflect.Type;
 import java.lang.reflect.UndeclaredThrowableException;
 import java.util.Collections;
 import java.util.HashMap;
@@ -109,7 +107,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 
 	private final Map<Method, MethodInvocationGateway> gatewayMap = new HashMap<>();
 
-	private Class<?> serviceInterface;
+	private Class<?> serviceInterface = RequestReplyExchanger.class;
 
 	private MessageChannel defaultRequestChannel;
 
@@ -159,7 +157,6 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 	 * {@link RequestReplyExchanger}, upon initialization.
 	 */
 	public GatewayProxyFactoryBean() {
-		// serviceInterface will be determined on demand later
 	}
 
 	public GatewayProxyFactoryBean(Class<?> serviceInterface) {
@@ -172,9 +169,10 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 	/**
 	 * Set the interface class that the generated proxy should implement.
 	 * If none is provided explicitly, the default is {@link RequestReplyExchanger}.
-	 *
 	 * @param serviceInterface The service interface.
+	 * @deprecated since 5.2.1 in favor of ctor initialization
 	 */
+	@Deprecated
 	public void setServiceInterface(Class<?> serviceInterface) {
 		Assert.notNull(serviceInterface, "'serviceInterface' must not be null");
 		Assert.isTrue(serviceInterface.isInterface(), "'serviceInterface' must be an interface");
@@ -247,7 +245,6 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 	/**
 	 * Set the default timeout value for sending request messages. If not explicitly
 	 * configured with an annotation, or on a method element, this value will be used.
-	 *
 	 * @param defaultRequestTimeout the timeout value in milliseconds
 	 */
 	public void setDefaultRequestTimeout(Long defaultRequestTimeout) {
@@ -258,7 +255,6 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 	 * Set an expression to be evaluated to determine the default timeout value for
 	 * sending request messages. If not explicitly configured with an annotation, or on a
 	 * method element, this value will be used.
-	 *
 	 * @param defaultRequestTimeout the timeout value in milliseconds
 	 * @since 5.0
 	 */
@@ -270,7 +266,6 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 	 * Set an expression to be evaluated to determine the default timeout value for
 	 * sending request messages. If not explicitly configured with an annotation, or on a
 	 * method element, this value will be used.
-	 *
 	 * @param defaultRequestTimeout the timeout value in milliseconds
 	 * @since 5.0
 	 */
@@ -283,7 +278,6 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 	/**
 	 * Set the default timeout value for receiving reply messages. If not explicitly
 	 * configured with an annotation, or on a method element, this value will be used.
-	 *
 	 * @param defaultReplyTimeout the timeout value in milliseconds
 	 */
 	public void setDefaultReplyTimeout(Long defaultReplyTimeout) {
@@ -294,7 +288,6 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 	 * Set an expression to be evaluated to determine the default timeout value for
 	 * receiving reply messages. If not explicitly configured with an annotation, or on a
 	 * method element, this value will be used.
-	 *
 	 * @param defaultReplyTimeout the timeout value in milliseconds
 	 * @since 5.0
 	 */
@@ -306,7 +299,6 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 	 * Set an expression to be evaluated to determine the default timeout value for
 	 * receiving reply messages. If not explicitly configured with an annotation, or on a
 	 * method element, this value will be used.
-	 *
 	 * @param defaultReplyTimeout the timeout value in milliseconds
 	 * @since 5.0
 	 */
@@ -389,18 +381,18 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 			if (this.initialized) {
 				return;
 			}
-			BeanFactory beanFactory = this.getBeanFactory();
+			BeanFactory beanFactory = getBeanFactory();
 			if (this.channelResolver == null && beanFactory != null) {
 				this.channelResolver = ChannelResolverUtils.getChannelResolver(beanFactory);
 			}
-			Class<?> proxyInterface = determineServiceInterface();
-			Method[] methods = ReflectionUtils.getUniqueDeclaredMethods(proxyInterface);
+			Method[] methods = ReflectionUtils.getUniqueDeclaredMethods(this.serviceInterface);
 			for (Method method : methods) {
 				MethodInvocationGateway gateway = createGatewayForMethod(method);
 				this.gatewayMap.put(method, gateway);
 			}
-			this.serviceProxy = new ProxyFactory(proxyInterface, this)
-					.getProxy(this.beanClassLoader);
+			this.serviceProxy =
+					new ProxyFactory(this.serviceInterface, this)
+							.getProxy(this.beanClassLoader);
 			if (this.asyncExecutor != null) {
 				Callable<String> task = () -> null;
 				Future<String> submitType = this.asyncExecutor.submit(task);
@@ -410,21 +402,14 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 					this.asyncSubmitListenableType = submitType.getClass();
 				}
 			}
-			this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
+			this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(beanFactory);
 			this.initialized = true;
 		}
 	}
 
-	private Class<?> determineServiceInterface() {
-		if (this.serviceInterface == null) {
-			this.serviceInterface = RequestReplyExchanger.class;
-		}
-		return this.serviceInterface;
-	}
-
 	@Override
 	public Class<?> getObjectType() {
-		return (this.serviceInterface != null ? this.serviceInterface : null);
+		return this.serviceInterface;
 	}
 
 	@Override
@@ -436,15 +421,14 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 		return this.serviceProxy;
 	}
 
-	@Override
-	public boolean isSingleton() {
-		return true;
-	}
-
 	@Override
 	@Nullable
 	public Object invoke(final MethodInvocation invocation) throws Throwable { // NOSONAR
-		final Class<?> returnType = invocation.getMethod().getReturnType();
+		Class<?> returnType = invocation.getMethod().getReturnType();
+		MethodInvocationGateway gateway = this.gatewayMap.get(invocation.getMethod());
+		if (gateway != null) {
+			returnType = gateway.returnType;
+		}
 		if (this.asyncExecutor != null && !Object.class.equals(returnType)) {
 			Invoker invoker = new Invoker(invocation);
 			if (returnType.isAssignableFrom(this.asyncSubmitType)) {
@@ -463,9 +447,11 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 			}
 		}
 		if (Mono.class.isAssignableFrom(returnType)) {
-			return Mono.fromSupplier(new Invoker(invocation));
+			return doInvoke(invocation, false);
+		}
+		else {
+			return doInvoke(invocation, true);
 		}
-		return doInvoke(invocation, true);
 	}
 
 	@Nullable
@@ -490,14 +476,9 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 		}
 		Method method = invocation.getMethod();
 		MethodInvocationGateway gateway = this.gatewayMap.get(method);
-		Class<?> returnType = method.getReturnType();
-		if (gateway.getReturnTypeMessage() == null) {
-			gateway.setReturnTypeMessage(Message.class.isAssignableFrom(returnType)
-					|| hasReturnMessageTypeOnFunction(method));
-		}
 		boolean shouldReturnMessage =
-				gateway.isReturnTypeMessage || hasReturnParameterizedWithMessage(method, runningOnCallerThread);
-		boolean shouldReply = returnType != void.class;
+				Message.class.isAssignableFrom(gateway.returnType) || (!runningOnCallerThread && gateway.expectMessage);
+		boolean shouldReply = gateway.returnType != void.class;
 		int paramCount = method.getParameterTypes().length;
 		Object response;
 		boolean hasPayloadExpression = findPayloadExpression(method);
@@ -507,7 +488,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 		else {
 			response = sendOrSendAndReceive(invocation, gateway, shouldReturnMessage, shouldReply);
 		}
-		return response(returnType, shouldReturnMessage, response);
+		return response(gateway.returnType, shouldReturnMessage, response);
 	}
 
 	@Nullable
@@ -567,16 +548,26 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 	@Nullable
 	private Object sendOrSendAndReceive(MethodInvocation invocation, MethodInvocationGateway gateway,
 			boolean shouldReturnMessage, boolean shouldReply) {
-		Object response;
+
 		Object[] args = invocation.getArguments();
 		if (shouldReply) {
-			response = shouldReturnMessage ? gateway.sendAndReceiveMessage(args) : gateway.sendAndReceive(args);
+			if (gateway.isMonoReturn) {
+				Mono<Message<?>> messageMono = gateway.sendAndReceiveMessageReactive(args);
+				if (!shouldReturnMessage) {
+					return messageMono.map(Message::getPayload);
+				}
+				else {
+					return messageMono;
+				}
+			}
+			else {
+				return shouldReturnMessage ? gateway.sendAndReceiveMessage(args) : gateway.sendAndReceive(args);
+			}
 		}
 		else {
 			gateway.send(args);
-			response = null;
 		}
-		return response;
+		return null;
 	}
 
 	private void rethrowExceptionCauseIfPossible(Throwable originalException, Method method)
@@ -618,7 +609,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 			annotationHeaders(gatewayAnnotation, headerExpressions);
 		}
 		else if (methodMetadata != null && !CollectionUtils.isEmpty(methodMetadata.getHeaderExpressions())) {
-				headerExpressions.putAll(methodMetadata.getHeaderExpressions());
+			headerExpressions.putAll(methodMetadata.getHeaderExpressions());
 		}
 
 		return doCreateMethodInvocationGateway(method, payloadExpression, headerExpressions,
@@ -767,6 +758,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 
 		GatewayMethodInboundMessageMapper messageMapper = createGatewayMessageMapper(method, headerExpressions);
 		MethodInvocationGateway gateway = new MethodInvocationGateway(messageMapper);
+		gateway.setupReturnType(this.serviceInterface, method);
 
 		JavaUtils.INSTANCE
 				.acceptIfNotNull(payloadExpression, messageMapper::setPayloadExpression)
@@ -784,8 +776,8 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 		return gateway;
 	}
 
-	private GatewayMethodInboundMessageMapper createGatewayMessageMapper(Method method, Map<String,
-			Expression> headerExpressions) {
+	private GatewayMethodInboundMessageMapper createGatewayMessageMapper(Method method,
+			Map<String, Expression> headerExpressions) {
 
 		Map<String, Object> headers = headers(method, headerExpressions);
 
@@ -878,8 +870,9 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 		}
 	}
 
-	private void setChannel(MessageChannel channel, Consumer<MessageChannel> channelMethod, String channelName,
-			Consumer<String> channelNameMethod) {
+	private void setChannel(@Nullable MessageChannel channel, Consumer<MessageChannel> channelMethod,
+			String channelName, Consumer<String> channelNameMethod) {
+
 		if (channel != null) {
 			channelMethod.accept(channel);
 		}
@@ -935,44 +928,15 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 		}
 	}
 
-	private static boolean hasReturnParameterizedWithMessage(Method method, boolean runningOnCallerThread) {
-		if (!runningOnCallerThread &&
-				(Future.class.isAssignableFrom(method.getReturnType())
-						|| Mono.class.isAssignableFrom(method.getReturnType()))) {
-			Type returnType = method.getGenericReturnType();
-			if (returnType instanceof ParameterizedType) {
-				Type[] typeArgs = ((ParameterizedType) returnType).getActualTypeArguments();
-				if (typeArgs != null && typeArgs.length == 1) {
-					Type parameterizedType = typeArgs[0];
-					if (parameterizedType instanceof ParameterizedType) {
-						Type rawType = ((ParameterizedType) parameterizedType).getRawType();
-						if (rawType instanceof Class) {
-							return Message.class.isAssignableFrom((Class<?>) rawType);
-						}
-					}
-				}
-			}
-		}
-		return false;
-	}
-
-	private boolean hasReturnMessageTypeOnFunction(Method method) {
-		if (Function.class.isAssignableFrom(this.serviceInterface) && "apply".equals(method.getName())) {
-			Class<?> returnType =
-					ResolvableType.forClass(Function.class, this.serviceInterface)
-							.getGeneric(1)
-							.getRawClass();
-			return returnType != null && Message.class.isAssignableFrom(returnType);
-		}
-		return false;
-	}
-
-
 	private static final class MethodInvocationGateway extends MessagingGatewaySupport {
 
 		private Expression receiveTimeoutExpression;
 
-		private volatile Boolean isReturnTypeMessage;
+		private Class<?> returnType;
+
+		private boolean expectMessage;
+
+		private boolean isMonoReturn;
 
 		MethodInvocationGateway(GatewayMethodInboundMessageMapper messageMapper) {
 			setRequestMapper(messageMapper);
@@ -987,13 +951,27 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 			this.receiveTimeoutExpression = receiveTimeoutExpression;
 		}
 
-		@Nullable
-		Boolean getReturnTypeMessage() {
-			return this.isReturnTypeMessage;
+		void setupReturnType(Class<?> serviceInterface, Method method) {
+			ResolvableType resolvableType;
+			if (Function.class.isAssignableFrom(serviceInterface) && "apply".equals(method.getName())) {
+				resolvableType = ResolvableType.forClass(Function.class, serviceInterface).getGeneric(1);
+			}
+			else {
+				resolvableType = ResolvableType.forMethodReturnType(method);
+			}
+			this.returnType = resolvableType.getRawClass();
+			if (this.returnType == null) {
+				this.returnType = Object.class;
+			}
+			else {
+				this.isMonoReturn = Mono.class.isAssignableFrom(this.returnType);
+				this.expectMessage = hasReturnParameterizedWithMessage(resolvableType);
+			}
 		}
 
-		void setReturnTypeMessage(Boolean returnTypeMessage) {
-			this.isReturnTypeMessage = returnTypeMessage;
+		private boolean hasReturnParameterizedWithMessage(ResolvableType resolvableType) {
+			return (Future.class.isAssignableFrom(this.returnType) || Mono.class.isAssignableFrom(this.returnType))
+					&& Message.class.isAssignableFrom(resolvableType.getGeneric(0).resolve(Object.class));
 		}
 
 	}
@@ -1007,6 +985,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 		}
 
 		@Override
+		@Nullable
 		public Object get() {
 			try {
 				return doInvoke(this.invocation, false);
@@ -1018,7 +997,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 				if (t instanceof RuntimeException) { //NOSONAR
 					throw (RuntimeException) t;
 				}
-				throw new MessagingException("Asynchronous gateway invocation failed", t);
+				throw new MessagingException("Asynchronous gateway invocation failed for: " + this.invocation, t);
 			}
 		}
 

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

@@ -17,9 +17,9 @@
 package org.springframework.integration.gateway;
 
 import java.util.Map;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicLong;
 
+import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 
 import org.springframework.beans.factory.BeanFactory;
@@ -59,6 +59,7 @@ import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.Assert;
 
 import reactor.core.publisher.Mono;
+import reactor.core.publisher.MonoProcessor;
 
 /**
  * A convenient base class for connecting application code to
@@ -74,7 +75,7 @@ import reactor.core.publisher.Mono;
 @IntegrationManagedResource
 public abstract class MessagingGatewaySupport extends AbstractEndpoint
 		implements org.springframework.integration.support.management.TrackableComponent,
-			org.springframework.integration.support.management.MessageSourceMetrics {
+		org.springframework.integration.support.management.MessageSourceMetrics {
 
 	private static final long DEFAULT_TIMEOUT = 1000L;
 
@@ -111,8 +112,6 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 
 	private InboundMessageMapper<Object> requestMapper = new DefaultRequestMapper();
 
-	private volatile AbstractEndpoint replyMessageCorrelator;
-
 	private String managedType;
 
 	private String managedName;
@@ -121,6 +120,8 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 
 	private boolean loggingEnabled = true;
 
+	private volatile AbstractEndpoint replyMessageCorrelator;
+
 	private volatile boolean initialized;
 
 
@@ -136,7 +137,6 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 	 * {@link ErrorMessage} with a {@link MessageTimeoutException} payload to the error
 	 * channel if a reply is expected but none is received. If no error channel is
 	 * configured, the {@link MessageTimeoutException} will be thrown.
-	 *
 	 * @param errorOnTimeout true to create the error message.
 	 * @since 4.2
 	 */
@@ -405,7 +405,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 	}
 
 	protected void send(Object object) {
-		this.initializeIfNecessary();
+		initializeIfNecessary();
 		Assert.notNull(object, "request must not be null");
 		MessageChannel channel = getRequestChannel();
 		Assert.state(channel != null,
@@ -429,7 +429,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 
 	@Nullable
 	protected Object receive() {
-		this.initializeIfNecessary();
+		initializeIfNecessary();
 		MessageChannel channel = getReplyChannel();
 		assertPollableChannel(channel);
 		return this.messagingTemplate.receiveAndConvert(channel, Object.class);
@@ -445,7 +445,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 
 	@Nullable
 	protected Object receive(long timeout) {
-		this.initializeIfNecessary();
+		initializeIfNecessary();
 		MessageChannel channel = getReplyChannel();
 		assertPollableChannel(channel);
 		return this.messagingTemplate.receiveAndConvert(channel, timeout);
@@ -612,7 +612,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 			Object originalReplyChannelHeader = message.getHeaders().getReplyChannel();
 			Object originalErrorChannelHeader = message.getHeaders().getErrorChannel();
 
-			FutureReplyChannel replyChan = new FutureReplyChannel();
+			MonoReplyChannel replyChan = new MonoReplyChannel();
 
 			Message<?> requestMessage = MutableMessageBuilder.fromMessage(message)
 					.setReplyChannel(replyChan)
@@ -623,7 +623,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 
 			sendMessageForReactiveFlow(requestChannel, requestMessage);
 
-			return buildReplyMono(requestMessage, replyChan, error, originalReplyChannelHeader,
+			return buildReplyMono(requestMessage, replyChan.replyMono, error, originalReplyChannelHeader,
 					originalErrorChannelHeader);
 		});
 	}
@@ -649,10 +649,10 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 		}
 	}
 
-	private Mono<Message<?>> buildReplyMono(Message<?> requestMessage, FutureReplyChannel replyChannel, boolean error,
+	private Mono<Message<?>> buildReplyMono(Message<?> requestMessage, Mono<Message<?>> reply, boolean error,
 			@Nullable Object originalReplyChannelHeader, @Nullable Object originalErrorChannelHeader) {
 
-		return Mono.fromFuture(replyChannel.messageFuture)
+		return reply
 				.doOnSubscribe(s -> {
 					if (!error && this.countsEnabled) {
 						this.messageCount.incrementAndGet();
@@ -841,17 +841,25 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 
 	}
 
-	private static class FutureReplyChannel implements MessageChannel {
+	private static class MonoReplyChannel implements MessageChannel, ReactiveStreamsSubscribableChannel {
 
-		private final CompletableFuture<Message<?>> messageFuture = new CompletableFuture<>();
+		private final MonoProcessor<Message<?>> replyMono = MonoProcessor.create();
 
-		FutureReplyChannel() {
+		MonoReplyChannel() {
 			super();
 		}
 
 		@Override
 		public boolean send(Message<?> message, long timeout) {
-			return this.messageFuture.complete(message);
+			this.replyMono.onNext(message);
+			this.replyMono.onComplete();
+			return true;
+		}
+
+		@Override
+		public void subscribeTo(Publisher<? extends Message<?>> publisher) {
+			this.replyMono.switchIfEmpty(Mono.from(publisher));
+			this.replyMono.onComplete();
 		}
 
 	}

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

@@ -25,7 +25,6 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.Test;
@@ -50,6 +49,7 @@ import reactor.core.publisher.Mono;
  * @author Oleg Zhurakousky
  * @author Gary Russell
  * @author Artem Bilan
+ *
  * @since 2.0
  */
 public class AsyncGatewayTests {
@@ -58,9 +58,8 @@ public class AsyncGatewayTests {
 	public void futureWithMessageReturned() throws Exception {
 		QueueChannel requestChannel = new QueueChannel();
 		startResponder(requestChannel);
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
-		proxyFactory.setServiceInterface(TestEchoService.class);
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
 		proxyFactory.afterPropertiesSet();
@@ -82,9 +81,8 @@ public class AsyncGatewayTests {
 			}
 
 		};
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(channel);
-		proxyFactory.setServiceInterface(TestEchoService.class);
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
 		proxyFactory.afterPropertiesSet();
@@ -104,16 +102,15 @@ public class AsyncGatewayTests {
 		QueueChannel requestChannel = new QueueChannel();
 		addThreadEnricher(requestChannel);
 		startResponder(requestChannel);
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
-		proxyFactory.setServiceInterface(TestEchoService.class);
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = (TestEchoService) proxyFactory.getObject();
 		ListenableFuture<Message<?>> f = service.returnMessageListenable("foo");
 		long start = System.currentTimeMillis();
-		final AtomicReference<Message<?>> result = new AtomicReference<Message<?>>();
+		final AtomicReference<Message<?>> result = new AtomicReference<>();
 		final CountDownLatch latch = new CountDownLatch(1);
 		f.addCallback(new ListenableFutureCallback<Message<?>>() {
 
@@ -141,9 +138,8 @@ public class AsyncGatewayTests {
 		QueueChannel requestChannel = new QueueChannel();
 		addThreadEnricher(requestChannel);
 		startResponder(requestChannel);
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
-		proxyFactory.setServiceInterface(TestEchoService.class);
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
 		proxyFactory.afterPropertiesSet();
@@ -159,9 +155,8 @@ public class AsyncGatewayTests {
 		QueueChannel requestChannel = new QueueChannel();
 		addThreadEnricher(requestChannel);
 		startResponder(requestChannel);
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
-		proxyFactory.setServiceInterface(TestEchoService.class);
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
 
@@ -192,9 +187,8 @@ public class AsyncGatewayTests {
 	public void futureWithPayloadReturned() throws Exception {
 		QueueChannel requestChannel = new QueueChannel();
 		startResponder(requestChannel);
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
-		proxyFactory.setServiceInterface(TestEchoService.class);
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
 		proxyFactory.afterPropertiesSet();
@@ -209,9 +203,8 @@ public class AsyncGatewayTests {
 	public void futureWithWildcardReturned() throws Exception {
 		QueueChannel requestChannel = new QueueChannel();
 		startResponder(requestChannel);
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
-		proxyFactory.setServiceInterface(TestEchoService.class);
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
 		proxyFactory.afterPropertiesSet();
@@ -224,12 +217,11 @@ public class AsyncGatewayTests {
 
 
 	@Test
-	public void monoWithMessageReturned() throws Exception {
+	public void monoWithMessageReturned() {
 		QueueChannel requestChannel = new QueueChannel();
 		startResponder(requestChannel);
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
-		proxyFactory.setServiceInterface(TestEchoService.class);
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.afterPropertiesSet();
@@ -240,12 +232,11 @@ public class AsyncGatewayTests {
 	}
 
 	@Test
-	public void monoWithPayloadReturned() throws Exception {
+	public void monoWithPayloadReturned() {
 		QueueChannel requestChannel = new QueueChannel();
 		startResponder(requestChannel);
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
-		proxyFactory.setServiceInterface(TestEchoService.class);
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.afterPropertiesSet();
@@ -256,12 +247,11 @@ public class AsyncGatewayTests {
 	}
 
 	@Test
-	public void monoWithWildcardReturned() throws Exception {
+	public void monoWithWildcardReturned() {
 		QueueChannel requestChannel = new QueueChannel();
 		startResponder(requestChannel);
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
-		proxyFactory.setServiceInterface(TestEchoService.class);
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.afterPropertiesSet();
@@ -276,16 +266,15 @@ public class AsyncGatewayTests {
 	public void monoWithConsumer() throws Exception {
 		QueueChannel requestChannel = new QueueChannel();
 		startResponder(requestChannel);
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
-		proxyFactory.setServiceInterface(TestEchoService.class);
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = (TestEchoService) proxyFactory.getObject();
 		Mono<String> mono = service.returnStringPromise("foo");
 
-		final AtomicReference<String> result = new AtomicReference<String>();
+		final AtomicReference<String> result = new AtomicReference<>();
 		final CountDownLatch latch = new CountDownLatch(1);
 
 		mono.subscribe(s -> {
@@ -374,14 +363,13 @@ public class AsyncGatewayTests {
 		}
 
 		@Override
-		public String get() throws InterruptedException, ExecutionException {
-			return result;
+		public String get() {
+			return this.result;
 		}
 
 		@Override
-		public String get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException,
-				TimeoutException {
-			return result;
+		public String get(long timeout, TimeUnit unit) {
+			return this.result;
 		}
 
 	}

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

@@ -17,7 +17,8 @@
 package org.springframework.integration.gateway;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.fail;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 
@@ -67,12 +68,11 @@ import org.springframework.util.ReflectionUtils;
 public class GatewayProxyFactoryBeanTests {
 
 	@Test
-	public void testRequestReplyWithAnonymousChannel() throws Exception {
+	public void testRequestReplyWithAnonymousChannel() {
 		QueueChannel requestChannel = new QueueChannel();
 		startResponder(requestChannel);
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
-		proxyFactory.setServiceInterface(TestService.class);
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.afterPropertiesSet();
@@ -82,7 +82,7 @@ public class GatewayProxyFactoryBeanTests {
 	}
 
 	@Test
-	public void testRequestReplyWithAnonymousChannelConvertedTypeViaConversionService() throws Exception {
+	public void testRequestReplyWithAnonymousChannelConvertedTypeViaConversionService() {
 		QueueChannel requestChannel = new QueueChannel();
 		startResponder(requestChannel);
 		GenericConversionService cs = new DefaultConversionService();
@@ -98,13 +98,12 @@ public class GatewayProxyFactoryBeanTests {
 				};
 		stringToByteConverter = spy(stringToByteConverter);
 		cs.addConverter(stringToByteConverter);
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestService.class);
 		DefaultListableBeanFactory bf = new DefaultListableBeanFactory();
 		bf.registerSingleton(IntegrationUtils.INTEGRATION_CONVERSION_SERVICE_BEAN_NAME, cs);
 
 		proxyFactory.setBeanFactory(bf);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
-		proxyFactory.setServiceInterface(TestService.class);
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.afterPropertiesSet();
 		TestService service = (TestService) proxyFactory.getObject();
@@ -114,10 +113,9 @@ public class GatewayProxyFactoryBeanTests {
 	}
 
 	@Test
-	public void testOneWay() throws Exception {
+	public void testOneWay() {
 		final QueueChannel requestChannel = new QueueChannel();
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
-		proxyFactory.setServiceInterface(TestService.class);
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
@@ -130,11 +128,10 @@ public class GatewayProxyFactoryBeanTests {
 	}
 
 	@Test
-	public void testSolicitResponse() throws Exception {
+	public void testSolicitResponse() {
 		QueueChannel replyChannel = new QueueChannel();
 		replyChannel.send(new GenericMessage<>("foo"));
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
-		proxyFactory.setServiceInterface(TestService.class);
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestService.class);
 		proxyFactory.setDefaultRequestChannel(new DirectChannel());
 		proxyFactory.setDefaultReplyChannel(replyChannel);
 		proxyFactory.setBeanName("testGateway");
@@ -147,11 +144,10 @@ public class GatewayProxyFactoryBeanTests {
 	}
 
 	@Test
-	public void testReceiveMessage() throws Exception {
+	public void testReceiveMessage() {
 		QueueChannel replyChannel = new QueueChannel();
 		replyChannel.send(new GenericMessage<>("foo"));
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
-		proxyFactory.setServiceInterface(TestService.class);
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestService.class);
 		proxyFactory.setDefaultReplyChannel(replyChannel);
 
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
@@ -163,15 +159,14 @@ public class GatewayProxyFactoryBeanTests {
 	}
 
 	@Test
-	public void testRequestReplyWithTypeConversion() throws Exception {
+	public void testRequestReplyWithTypeConversion() {
 		final QueueChannel requestChannel = new QueueChannel();
 		new Thread(() -> {
 			Message<?> input = requestChannel.receive();
 			GenericMessage<String> reply = new GenericMessage<>(input.getPayload() + "456");
 			((MessageChannel) input.getHeaders().getReplyChannel()).send(reply);
 		}).start();
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
-		proxyFactory.setServiceInterface(TestService.class);
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
@@ -239,11 +234,10 @@ public class GatewayProxyFactoryBeanTests {
 	}
 
 	@Test
-	public void testMessageAsMethodArgument() throws Exception {
+	public void testMessageAsMethodArgument() {
 		QueueChannel requestChannel = new QueueChannel();
 		startResponder(requestChannel);
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
-		proxyFactory.setServiceInterface(TestService.class);
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
@@ -254,11 +248,10 @@ public class GatewayProxyFactoryBeanTests {
 	}
 
 	@Test
-	public void testNoArgMethodWithPayloadAnnotation() throws Exception {
+	public void testNoArgMethodWithPayloadAnnotation() {
 		QueueChannel requestChannel = new QueueChannel();
 		startResponder(requestChannel);
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
-		proxyFactory.setServiceInterface(TestService.class);
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
@@ -269,15 +262,14 @@ public class GatewayProxyFactoryBeanTests {
 	}
 
 	@Test
-	public void testMessageAsReturnValue() throws Exception {
+	public void testMessageAsReturnValue() {
 		final QueueChannel requestChannel = new QueueChannel();
 		new Thread(() -> {
 			Message<?> input = requestChannel.receive();
 			GenericMessage<String> reply = new GenericMessage<>(input.getPayload() + "bar");
 			((MessageChannel) input.getHeaders().getReplyChannel()).send(reply);
 		}).start();
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
-		proxyFactory.setServiceInterface(TestService.class);
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
@@ -289,25 +281,14 @@ public class GatewayProxyFactoryBeanTests {
 
 	@Test
 	public void testServiceMustBeInterface() {
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
-		int count = 0;
-		try {
-			proxyFactory.setServiceInterface(TestService.class);
-			count++;
-			proxyFactory.setServiceInterface(String.class);
-			count++;
-		}
-		catch (IllegalArgumentException e) {
-			// expected
-		}
-		assertThat(count).isEqualTo(1);
+		assertThatIllegalArgumentException()
+				.isThrownBy(() -> new GatewayProxyFactoryBean(String.class));
 	}
 
 	@Test
-	public void testProxiedToStringMethod() throws Exception {
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
+	public void testProxiedToStringMethod() {
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestService.class);
 		proxyFactory.setDefaultRequestChannel(new DirectChannel());
-		proxyFactory.setServiceInterface(TestService.class);
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
 		proxyFactory.afterPropertiesSet();
@@ -316,9 +297,9 @@ public class GatewayProxyFactoryBeanTests {
 		assertThat(proxy.toString().substring(0, expected.length())).isEqualTo(expected);
 	}
 
-	@Test(expected = TestException.class)
-	public void testCheckedExceptionRethrownAsIs() throws Exception {
-		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
+	@Test
+	public void testCheckedExceptionRethrownAsIs() {
+		GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestExceptionThrowingInterface.class);
 		DirectChannel channel = new DirectChannel();
 		EventDrivenConsumer consumer = new EventDrivenConsumer(channel, new MessageHandler() {
 
@@ -331,12 +312,12 @@ public class GatewayProxyFactoryBeanTests {
 		});
 		consumer.start();
 		proxyFactory.setDefaultRequestChannel(channel);
-		proxyFactory.setServiceInterface(TestExceptionThrowingInterface.class);
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.setBeanFactory(mock(BeanFactory.class));
 		proxyFactory.afterPropertiesSet();
 		TestExceptionThrowingInterface proxy = (TestExceptionThrowingInterface) proxyFactory.getObject();
-		proxy.throwCheckedException("test");
+		assertThatExceptionOfType(TestException.class)
+				.isThrownBy(() -> proxy.throwCheckedException("test"));
 	}
 
 
@@ -349,10 +330,9 @@ public class GatewayProxyFactoryBeanTests {
 	}
 
 	@Test
-	public void testProgrammaticWiring() throws Exception {
-		GatewayProxyFactoryBean gpfb = new GatewayProxyFactoryBean();
+	public void testProgrammaticWiring() {
+		GatewayProxyFactoryBean gpfb = new GatewayProxyFactoryBean(TestEchoService.class);
 		gpfb.setBeanFactory(mock(BeanFactory.class));
-		gpfb.setServiceInterface(TestEchoService.class);
 		QueueChannel drc = new QueueChannel();
 		gpfb.setDefaultRequestChannel(drc);
 		gpfb.setDefaultReplyTimeout(0L);
@@ -377,49 +357,29 @@ public class GatewayProxyFactoryBeanTests {
 		meta.setHeaderExpressions(Collections.singletonMap(MessageHeaders.ID, new LiteralExpression("bar")));
 		gpfb.setGlobalMethodMetadata(meta);
 
-		try {
-			gpfb.afterPropertiesSet();
-			fail("BeanInitializationException expected");
-		}
-		catch (Exception e) {
-			assertThat(e).isInstanceOf(BeanInitializationException.class);
-			assertThat(e.getMessage())
-					.contains("Messaging Gateway cannot override 'id' and 'timestamp' read-only headers");
-		}
+		assertThatExceptionOfType(BeanInitializationException.class)
+				.isThrownBy(gpfb::afterPropertiesSet)
+				.withMessageContaining("Messaging Gateway cannot override 'id' and 'timestamp' read-only headers");
 	}
 
 	@Test
 	public void testIdHeaderOverrideGatewayHeaderAnnotation() {
-		GatewayProxyFactoryBean gpfb = new GatewayProxyFactoryBean();
+		GatewayProxyFactoryBean gpfb = new GatewayProxyFactoryBean(HeadersOverwriteService.class);
 		gpfb.setBeanFactory(mock(BeanFactory.class));
-		gpfb.setServiceInterface(HeadersOverwriteService.class);
 
-		try {
-			gpfb.afterPropertiesSet();
-			fail("BeanInitializationException expected");
-		}
-		catch (Exception e) {
-			assertThat(e).isInstanceOf(BeanInitializationException.class);
-			assertThat(e.getMessage())
-					.contains("Messaging Gateway cannot override 'id' and 'timestamp' read-only headers");
-		}
+		assertThatExceptionOfType(BeanInitializationException.class)
+				.isThrownBy(gpfb::afterPropertiesSet)
+				.withMessageContaining("Messaging Gateway cannot override 'id' and 'timestamp' read-only headers");
 	}
 
 	@Test
 	public void testTimeStampHeaderOverrideParamHeaderAnnotation() {
-		GatewayProxyFactoryBean gpfb = new GatewayProxyFactoryBean();
+		GatewayProxyFactoryBean gpfb = new GatewayProxyFactoryBean(HeadersParamService.class);
 		gpfb.setBeanFactory(mock(BeanFactory.class));
-		gpfb.setServiceInterface(HeadersParamService.class);
 
-		try {
-			gpfb.afterPropertiesSet();
-			fail("BeanInitializationException expected");
-		}
-		catch (Exception e) {
-			assertThat(e).isInstanceOf(BeanInitializationException.class);
-			assertThat(e.getMessage())
-					.contains("Messaging Gateway cannot override 'id' and 'timestamp' read-only headers");
-		}
+		assertThatExceptionOfType(BeanInitializationException.class)
+				.isThrownBy(gpfb::afterPropertiesSet)
+				.withMessageContaining("Messaging Gateway cannot override 'id' and 'timestamp' read-only headers");
 	}
 
 	//	@Test

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

@@ -17,6 +17,7 @@
 package org.springframework.integration.gateway;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -39,6 +40,7 @@ import org.springframework.messaging.handler.annotation.Payload;
  * @author Mark Fisher
  * @author Gary Russell
  * @author Artem Bilan
+ *
  * @since 2.0
  */
 public class GatewayProxyMessageMappingTests {
@@ -49,9 +51,8 @@ public class GatewayProxyMessageMappingTests {
 
 
 	@Before
-	public void initializeGateway() throws Exception {
-		GatewayProxyFactoryBean factoryBean = new GatewayProxyFactoryBean();
-		factoryBean.setServiceInterface(TestGateway.class);
+	public void initializeGateway() {
+		GatewayProxyFactoryBean factoryBean = new GatewayProxyFactoryBean(TestGateway.class);
 		factoryBean.setDefaultRequestChannel(channel);
 		factoryBean.setBeanName("testGateway");
 		GenericApplicationContext context = new GenericApplicationContext();
@@ -65,8 +66,8 @@ public class GatewayProxyMessageMappingTests {
 
 
 	@Test
-	public void payloadAndHeaderMapWithoutAnnotations() throws Exception {
-		Map<String, Object> m = new HashMap<String, Object>();
+	public void payloadAndHeaderMapWithoutAnnotations() {
+		Map<String, Object> m = new HashMap<>();
 		m.put("k1", "v1");
 		m.put("k2", "v2");
 		gateway.payloadAndHeaderMapWithoutAnnotations("foo", m);
@@ -78,8 +79,8 @@ public class GatewayProxyMessageMappingTests {
 	}
 
 	@Test
-	public void payloadAndHeaderMapWithAnnotations() throws Exception {
-		Map<String, Object> m = new HashMap<String, Object>();
+	public void payloadAndHeaderMapWithAnnotations() {
+		Map<String, Object> m = new HashMap<>();
 		m.put("k1", "v1");
 		m.put("k2", "v2");
 		gateway.payloadAndHeaderMapWithAnnotations("foo", m);
@@ -91,7 +92,7 @@ public class GatewayProxyMessageMappingTests {
 	}
 
 	@Test
-	public void headerValuesAndPayloadWithAnnotations() throws Exception {
+	public void headerValuesAndPayloadWithAnnotations() {
 		gateway.headerValuesAndPayloadWithAnnotations("headerValue1", "payloadValue", "headerValue2");
 		Message<?> result = channel.receive(0);
 		assertThat(result).isNotNull();
@@ -101,8 +102,8 @@ public class GatewayProxyMessageMappingTests {
 	}
 
 	@Test
-	public void mapOnly() throws Exception {
-		Map<String, Object> map = new HashMap<String, Object>();
+	public void mapOnly() {
+		Map<String, Object> map = new HashMap<>();
 		map.put("k1", "v1");
 		map.put("k2", "v2");
 		gateway.mapOnly(map);
@@ -115,8 +116,8 @@ public class GatewayProxyMessageMappingTests {
 
 	@Test
 	public void twoMapsAndOneAnnotatedWithPayload() {
-		Map<String, Object> map1 = new HashMap<String, Object>();
-		Map<String, Object> map2 = new HashMap<String, Object>();
+		Map<String, Object> map1 = new HashMap<>();
+		Map<String, Object> map2 = new HashMap<>();
 		map1.put("k1", "v1");
 		map2.put("k2", "v2");
 		gateway.twoMapsAndOneAnnotatedWithPayload(map1, map2);
@@ -128,7 +129,7 @@ public class GatewayProxyMessageMappingTests {
 	}
 
 	@Test
-	public void payloadAnnotationAtMethodLevel() throws Exception {
+	public void payloadAnnotationAtMethodLevel() {
 		gateway.payloadAnnotationAtMethodLevel("foo", "bar");
 		Message<?> result = channel.receive(0);
 		assertThat(result).isNotNull();
@@ -136,7 +137,7 @@ public class GatewayProxyMessageMappingTests {
 	}
 
 	@Test
-	public void payloadAnnotationAtMethodLevelUsingBeanResolver() throws Exception {
+	public void payloadAnnotationAtMethodLevelUsingBeanResolver() {
 		GenericApplicationContext context = new GenericApplicationContext();
 		RootBeanDefinition gatewayDefinition = new RootBeanDefinition(GatewayProxyFactoryBean.class);
 		gatewayDefinition.getPropertyValues().add("defaultRequestChannel", channel);
@@ -155,7 +156,7 @@ public class GatewayProxyMessageMappingTests {
 	}
 
 	@Test
-	public void payloadAnnotationWithExpression() throws Exception {
+	public void payloadAnnotationWithExpression() {
 		gateway.payloadAnnotationWithExpression("foo");
 		Message<?> result = channel.receive(0);
 		assertThat(result).isNotNull();
@@ -163,7 +164,7 @@ public class GatewayProxyMessageMappingTests {
 	}
 
 	@Test
-	public void payloadAnnotationWithExpressionUsingBeanResolver() throws Exception {
+	public void payloadAnnotationWithExpressionUsingBeanResolver() {
 		GenericApplicationContext context = new GenericApplicationContext();
 		RootBeanDefinition gatewayDefinition = new RootBeanDefinition(GatewayProxyFactoryBean.class);
 		gatewayDefinition.getPropertyValues().add("defaultRequestChannel", channel);
@@ -186,28 +187,32 @@ public class GatewayProxyMessageMappingTests {
 		context.close();
 	}
 
-	@Test(expected = MessagingException.class)
+	@Test
 	public void twoMapsWithoutAnnotations() {
-		Map<String, Object> map1 = new HashMap<String, Object>();
-		Map<String, Object> map2 = new HashMap<String, Object>();
+		Map<String, Object> map1 = new HashMap<>();
+		Map<String, Object> map2 = new HashMap<>();
 		map1.put("k1", "v1");
 		map2.put("k2", "v2");
-		gateway.twoMapsWithoutAnnotations(map1, map2);
+		assertThatExceptionOfType(MessagingException.class)
+				.isThrownBy(() -> this.gateway.twoMapsWithoutAnnotations(map1, map2));
 	}
 
-	@Test(expected = MessagingException.class)
-	public void twoPayloads() throws Exception {
-		gateway.twoPayloads("won't", "work");
+	@Test
+	public void twoPayloads() {
+		assertThatExceptionOfType(MessagingException.class)
+				.isThrownBy(() -> this.gateway.twoPayloads("won't", "work"));
 	}
 
-	@Test(expected = MessagingException.class)
-	public void payloadAndHeaderAnnotationsOnSameParameter() throws Exception {
-		gateway.payloadAndHeaderAnnotationsOnSameParameter("oops");
+	@Test
+	public void payloadAndHeaderAnnotationsOnSameParameter() {
+		assertThatExceptionOfType(MessagingException.class)
+				.isThrownBy(() -> this.gateway.payloadAndHeaderAnnotationsOnSameParameter("oops"));
 	}
 
-	@Test(expected = MessagingException.class)
-	public void payloadAndHeadersAnnotationsOnSameParameter() throws Exception {
-		gateway.payloadAndHeadersAnnotationsOnSameParameter(new HashMap<String, Object>());
+	@Test
+	public void payloadAndHeadersAnnotationsOnSameParameter() {
+		assertThatExceptionOfType(MessagingException.class)
+				.isThrownBy(() -> this.gateway.payloadAndHeadersAnnotationsOnSameParameter(new HashMap<>()));
 	}
 
 
@@ -261,6 +266,7 @@ public class GatewayProxyMessageMappingTests {
 			}
 			return sum;
 		}
+
 	}
 
 }

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

@@ -14,7 +14,7 @@
 	<channel id="requestChannel"/>
 
 	<beans:bean id="proxy" class="org.springframework.integration.gateway.GatewayProxyFactoryBean">
-		<beans:property name="serviceInterface" value="org.springframework.integration.gateway.TestService"/>
+		<beans:constructor-arg value="org.springframework.integration.gateway.TestService"/>
 		<beans:property name="defaultRequestChannel" ref="requestChannel"/>
 	</beans:bean>
 

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

@@ -12,7 +12,7 @@
 	<service-activator ref="testHandler" input-channel="requestChannel"/>
 
 	<beans:bean id="proxy" class="org.springframework.integration.gateway.GatewayProxyFactoryBean">
-		<beans:property name="serviceInterface" value="org.springframework.integration.gateway.TestService"/>
+		<beans:constructor-arg value="org.springframework.integration.gateway.TestService"/>
 		<beans:property name="defaultRequestChannel" ref="requestChannel"/>
 	</beans:bean>
 

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

@@ -19,7 +19,7 @@
 	<service-activator ref="handler" input-channel="requestChannel" output-channel="replyChannel"/>
 
 	<beans:bean id="proxy" class="org.springframework.integration.gateway.GatewayProxyFactoryBean">
-		<beans:property name="serviceInterface" value="org.springframework.integration.gateway.TestService"/>
+		<beans:constructor-arg value="org.springframework.integration.gateway.TestService"/>
 		<beans:property name="defaultRequestChannel" ref="requestChannel"/>
 		<beans:property name="defaultReplyChannel" ref="replyChannel"/>
 		<beans:property name="defaultRequestTimeout" value="10000"/>

spring-integration-core/src/test/kotlin/org/springframework/integration/function/FunctionsTests.kt

@@ -44,9 +44,12 @@ import org.springframework.messaging.support.GenericMessage
 import org.springframework.messaging.support.MessageBuilder
 import org.springframework.test.annotation.DirtiesContext
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig
+import reactor.core.publisher.Mono
+import reactor.test.StepVerifier
 import java.util.*
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
+import java.util.function.Function
 
 /**
  * @author Artem Bilan
@@ -120,6 +123,18 @@ class FunctionsTests {
 		assertThat(this.fromSupplierQueue.receive(10_000)).isNotNull()
 	}
 
+	@Autowired
+	private lateinit var monoFunction: Function<String, Mono<Message<*>>>
+
+	@Test
+	fun `verify Mono gateway`() {
+		val mono = this.monoFunction.apply("test")
+
+		StepVerifier.create(mono.map(Message<*>::getPayload).cast(String::class.java))
+				.expectNext("TEST")
+				.verifyComplete()
+	}
+
 	@Configuration
 	@EnableIntegration
 	class Config {
@@ -155,6 +170,14 @@ class FunctionsTests {
 				IntegrationFlows.from<String>({ "bar" }) { e -> e.poller { p -> p.fixedDelay(10).maxMessagesPerPoll(1) } }
 						.channel { c -> c.queue("fromSupplierQueue") }
 						.get()
+
+		@Bean
+		fun monoFunctionGateway() =
+				IntegrationFlows.from(MonoFunction::class.java)
+						.handle<String>({ p, _ -> Mono.just(p).map(String::toUpperCase) }) { e -> e.async(true) }
+						.get()
 	}
 
+	interface MonoFunction : Function<String, Mono<Message<*>>>
+
 }