From 911cdc86b5f1dd8a2f3a612cbca0192ed79ee0ee Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 14 Oct 2019 15:31:04 -0400 Subject: [PATCH] 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 --- .../expression/ExpressionUtils.java | 12 +- .../gateway/GatewayMessageHandler.java | 1 - .../gateway/GatewayProxyFactoryBean.java | 159 ++++++++---------- .../gateway/MessagingGatewaySupport.java | 40 +++-- .../gateway/AsyncGatewayTests.java | 54 +++--- .../gateway/GatewayProxyFactoryBeanTests.java | 124 +++++--------- .../GatewayProxyMessageMappingTests.java | 64 +++---- .../integration/gateway/gatewayAutowiring.xml | 2 +- .../gateway/gatewayWithRendezvousChannel.xml | 2 +- .../gateway/gatewayWithResponseCorrelator.xml | 2 +- .../integration/function/FunctionsTests.kt | 23 +++ 11 files changed, 224 insertions(+), 259 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/expression/ExpressionUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/expression/ExpressionUtils.java index ef750bc7b9..b68b88b1b0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/expression/ExpressionUtils.java +++ b/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) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMessageHandler.java index e1950f539b..839ec77404 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMessageHandler.java +++ b/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) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java index d6cee539d1..120d26c2a0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java +++ b/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 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 task = () -> null; Future 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> 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 headerExpressions) { + private GatewayMethodInboundMessageMapper createGatewayMessageMapper(Method method, + Map headerExpressions) { Map headers = headers(method, headerExpressions); @@ -878,8 +870,9 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint } } - private void setChannel(MessageChannel channel, Consumer channelMethod, String channelName, - Consumer channelNameMethod) { + private void setChannel(@Nullable MessageChannel channel, Consumer channelMethod, + String channelName, Consumer 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); } } 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 624e5ff589..0b87b39218 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 @@ -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 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> buildReplyMono(Message requestMessage, FutureReplyChannel replyChannel, boolean error, + private Mono> buildReplyMono(Message requestMessage, Mono> 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> messageFuture = new CompletableFuture<>(); + private final MonoProcessor> 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> publisher) { + this.replyMono.switchIfEmpty(Mono.from(publisher)); + this.replyMono.onComplete(); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java index e3bdb1c5da..993549c47a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java +++ b/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> f = service.returnMessageListenable("foo"); long start = System.currentTimeMillis(); - final AtomicReference> result = new AtomicReference>(); + final AtomicReference> result = new AtomicReference<>(); final CountDownLatch latch = new CountDownLatch(1); f.addCallback(new ListenableFutureCallback>() { @@ -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 mono = service.returnStringPromise("foo"); - final AtomicReference result = new AtomicReference(); + final AtomicReference 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; } } 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 4dd61e7d54..e21ce82121 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 @@ -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 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 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 diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyMessageMappingTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyMessageMappingTests.java index d0f40039d9..f29d4c7811 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyMessageMappingTests.java +++ b/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 m = new HashMap(); + public void payloadAndHeaderMapWithoutAnnotations() { + Map 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 m = new HashMap(); + public void payloadAndHeaderMapWithAnnotations() { + Map 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 map = new HashMap(); + public void mapOnly() { + Map 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 map1 = new HashMap(); - Map map2 = new HashMap(); + Map map1 = new HashMap<>(); + Map 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 map1 = new HashMap(); - Map map2 = new HashMap(); + Map map1 = new HashMap<>(); + Map 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()); + @Test + public void payloadAndHeadersAnnotationsOnSameParameter() { + assertThatExceptionOfType(MessagingException.class) + .isThrownBy(() -> this.gateway.payloadAndHeadersAnnotationsOnSameParameter(new HashMap<>())); } @@ -261,6 +266,7 @@ public class GatewayProxyMessageMappingTests { } return sum; } + } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayAutowiring.xml b/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayAutowiring.xml index ddbf123ca2..14f4726fcb 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayAutowiring.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayAutowiring.xml @@ -14,7 +14,7 @@ - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayWithRendezvousChannel.xml b/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayWithRendezvousChannel.xml index 2ecede78d6..f3a9bf1937 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayWithRendezvousChannel.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayWithRendezvousChannel.xml @@ -12,7 +12,7 @@ - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayWithResponseCorrelator.xml b/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayWithResponseCorrelator.xml index 639a0775d0..ad40619945 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayWithResponseCorrelator.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayWithResponseCorrelator.xml @@ -19,7 +19,7 @@ - + diff --git a/spring-integration-core/src/test/kotlin/org/springframework/integration/function/FunctionsTests.kt b/spring-integration-core/src/test/kotlin/org/springframework/integration/function/FunctionsTests.kt index 463484b7ae..fb01894a32 100644 --- a/spring-integration-core/src/test/kotlin/org/springframework/integration/function/FunctionsTests.kt +++ b/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>> + + @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({ "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({ p, _ -> Mono.just(p).map(String::toUpperCase) }) { e -> e.async(true) } + .get() } + interface MonoFunction : Function>> + }