From b3db6a97a25f397bc25c4868422c902035b540fb Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 12 Jan 2017 16:54:07 -0500 Subject: [PATCH] INT-4214: Migrate to `InvocableHandlerMethod` JIRA: https://jira.spring.io/browse/INT-4214 * Rework `MessagingMethodInvokerHelper` to delegate to the `InvocableHandlerMethod` with its `HandlerMethodArgumentResolver` infrastructure instead of SpEL * Introduce several `HandlerMethodArgumentResolver` to address some SI use-cases like `@Payloads`, `@Payload(expression = "")` and `Collection` as argument * Initialize `DefaultMessageHandlerMethodFactory` in the `MessagingMethodInvokerHelper.start()`. With that I observed several `Lifecycle` problem when we don't have proper delegate from the top. * Fix `AbstractCorrelatingMessageHandler` and similar to delegate `Lifecycle` properly * Fix `ReactiveConsumer` to delegate `Lifecycle` to the `MessageHandler` * Fix `MutableMessage` do not `generateId()` and set `timestamp` headers if we already have them in the provided headers * With all that `Lifecycle` many tests must be fixed to call `start()` Add SpEL fallback variant to the MessagingMethodInvokerHelper Add `MessagingMethodInvokerHelper.setUseSpelInvoker(boolean)` Add `MethodInvokingMessageProcessorTests.testPerformanceSpelVersusInvocable()` Add Compiled SpEL comparison * Add `MapArgumentResolver` to cover `Properties` case * add `MessagingMethodInvokerHelper.HandlerMethod.spelOnly` state, when we definitely can perform ony SpEL for provided arguments, e.g. `@Header` with expression * Catch `IllegalStateException` with the `"argument type mismatch"` message to fallback to SpEL invocation * Add `Iterator` support for the `CollectionArgumentResolver` * Add `integrationConversionService` bean registration into the `TestUtils.createTestApplicationContext()` * Tweak `.travis.yml` to try to download latest JDK, the current `1.8.0_31` is pretty old already and has some bugs * Adjust some failing tests to use `TestUtils.createTestApplicationContext()` to rely on newly added `integrationConversionService` bean Fix failed tests: `ctx.refresh()` Polishing current year in Copyright * JavaDocs for `CollectionArgumentResolver` and `MapArgumentResolver` * Propagate `ConversionService` from the `MessagingMethodInvokerHelper` to the `MapArgumentResolver.java` Fix `MessagingMethodInvokerHelper` to propagate `BeanFactory` into `MapArgumentResolver` as well The `MapArgumentResolver` now `extends AbstractExpressionEvaluator`, too Fix JavaDoc typo --- .travis.yml | 1 + .../AbstractCorrelatingMessageHandler.java | 40 ++- .../MethodInvokingCorrelationStrategy.java | 22 +- .../MethodInvokingMessageGroupProcessor.java | 23 +- .../MethodInvokingMessageListProcessor.java | 32 ++- .../MethodInvokingReleaseStrategy.java | 21 +- .../config/ConsumerEndpointFactoryBean.java | 11 +- ...AbstractMethodAnnotationPostProcessor.java | 18 +- .../endpoint/EventDrivenConsumer.java | 5 +- .../endpoint/ReactiveConsumer.java | 52 +++- .../MethodInvokingMessageProcessor.java | 13 +- .../support/CollectionArgumentResolver.java | 104 +++++++ .../handler/support/MapArgumentResolver.java | 89 ++++++ .../PayloadExpressionArgumentResolver.java | 62 ++++ .../support/PayloadsArgumentResolver.java | 93 ++++++ .../handler/support/package-info.java | 4 + .../integration/support/MutableMessage.java | 8 +- .../support/MutableMessageHeaders.java | 13 +- .../util/MessagingMethodInvokerHelper.java | 265 +++++++++++++----- ...atingMessageGroupProcessorHeaderTests.java | 13 +- ...hodInvokingMessageGroupProcessorTests.java | 53 +++- .../ApplicationContextMessageBusTests.java | 13 +- .../integration/bus/messageBusTests.xml | 2 + .../reactive/ReactiveConsumerTests.java | 2 +- .../annotation/AggregatorAnnotationTests.java | 25 +- .../annotation/SubscriberOrderTests.java | 10 +- .../endpoint/MessageProducerSupportTests.java | 12 +- .../ServiceActivatorEndpointTests.java | 12 +- ...FilterAnnotationMethodResolutionTests.java | 16 +- .../gateway/MessagingGatewayTests.java | 7 +- .../MethodInvokingMessageProcessorTests.java | 51 +++- .../message/PayloadAndHeaderMappingTests.java | 29 +- .../router/config/RouterFactoryBeanTests.java | 16 +- .../util/BeanFactoryTypeConverterTests.java | 37 ++- .../integration/test/util/TestUtils.java | 4 +- 35 files changed, 974 insertions(+), 204 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/handler/support/CollectionArgumentResolver.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/handler/support/MapArgumentResolver.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/handler/support/PayloadExpressionArgumentResolver.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/handler/support/PayloadsArgumentResolver.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/handler/support/package-info.java diff --git a/.travis.yml b/.travis.yml index 93a98e3b40..f1806a8756 100644 --- a/.travis.yml +++ b/.travis.yml @@ -11,6 +11,7 @@ addons: - mongodb-3.0-precise packages: - mongodb-org-server + - oracle-java8-installer before_cache: - rm -f $HOME/.gradle/caches/modules-2/modules-2.lock cache: diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index 5cbf688acb..a02ec2b022 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -37,6 +37,7 @@ import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.DisposableBean; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; +import org.springframework.context.Lifecycle; import org.springframework.expression.EvaluationContext; import org.springframework.expression.Expression; import org.springframework.integration.IntegrationMessageHeaderAccessor; @@ -85,7 +86,7 @@ import org.springframework.util.CollectionUtils; * @since 2.0 */ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageProducingHandler - implements DiscardingMessageHandler, DisposableBean, ApplicationEventPublisherAware { + implements DiscardingMessageHandler, DisposableBean, ApplicationEventPublisherAware, Lifecycle { protected final Log logger = LogFactory.getLog(getClass()); @@ -131,6 +132,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP private volatile boolean expireGroupsUponTimeout = true; + private volatile boolean running; + public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store, CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) { Assert.notNull(processor, "'processor' must not be null"); @@ -750,6 +753,37 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } } + @Override + public void start() { + if (!this.running) { + this.running = true; + if (this.outputProcessor instanceof Lifecycle) { + ((Lifecycle) this.outputProcessor).start(); + } + if (this.releaseStrategy instanceof Lifecycle) { + ((Lifecycle) this.releaseStrategy).start(); + } + } + } + + @Override + public void stop() { + if (this.running) { + this.running = false; + if (this.outputProcessor instanceof Lifecycle) { + ((Lifecycle) this.outputProcessor).stop(); + } + if (this.releaseStrategy instanceof Lifecycle) { + ((Lifecycle) this.releaseStrategy).stop(); + } + } + } + + @Override + public boolean isRunning() { + return this.running; + } + protected static class SequenceAwareMessageGroup extends SimpleMessageGroup { private final SimpleMessageGroup sourceGroup; @@ -787,7 +821,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP Integer messageSequenceSize = message.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, Integer.class); if (messageSequenceSize == null) { - messageSequenceSize = Integer.valueOf(0); + messageSequenceSize = 0; } return messageSequenceSize.equals(getSequenceSize()) && !(this.sourceGroup != null ? this.sourceGroup.containsSequence(messageSequenceNumber) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingCorrelationStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingCorrelationStrategy.java index f074a734ef..3eeed7a8da 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingCorrelationStrategy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingCorrelationStrategy.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,6 +21,7 @@ import java.lang.reflect.Method; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.context.Lifecycle; import org.springframework.integration.handler.MethodInvokingMessageProcessor; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -33,12 +34,12 @@ import org.springframework.util.Assert; * @author Artem Bilan * @author Gary Russell */ -public class MethodInvokingCorrelationStrategy implements CorrelationStrategy, BeanFactoryAware { +public class MethodInvokingCorrelationStrategy implements CorrelationStrategy, BeanFactoryAware, Lifecycle { private final MethodInvokingMessageProcessor processor; public MethodInvokingCorrelationStrategy(Object object, String methodName) { - this.processor = new MethodInvokingMessageProcessor(object, methodName, true); + this.processor = new MethodInvokingMessageProcessor(object, methodName); } public MethodInvokingCorrelationStrategy(Object object, Method method) { @@ -60,4 +61,19 @@ public class MethodInvokingCorrelationStrategy implements CorrelationStrategy, B return this.processor.processMessage(message); } + @Override + public void start() { + this.processor.start(); + } + + @Override + public void stop() { + this.processor.stop(); + } + + @Override + public boolean isRunning() { + return this.processor.isRunning(); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java index 0d3fdc44a7..b8939a6348 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,6 +21,7 @@ import java.util.Collection; import java.util.Map; import org.springframework.beans.factory.BeanFactory; +import org.springframework.context.Lifecycle; import org.springframework.core.convert.ConversionService; import org.springframework.integration.annotation.Aggregator; import org.springframework.integration.store.MessageGroup; @@ -33,9 +34,12 @@ import org.springframework.messaging.Message; * @author Mark Fisher * @author Dave Syer * @author Gary Russell + * @author Artme Bilan + * * @since 2.0 */ -public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMessageGroupProcessor { +public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMessageGroupProcessor + implements Lifecycle { private final MethodInvokingMessageListProcessor processor; @@ -86,4 +90,19 @@ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMess return this.processor.process(messagesUpForProcessing, headers); } + @Override + public void start() { + this.processor.start(); + } + + @Override + public void stop() { + this.processor.stop(); + } + + @Override + public boolean isRunning() { + return this.processor.isRunning(); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageListProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageListProcessor.java index 949efdb6d8..1716c8d774 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageListProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageListProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -22,6 +22,7 @@ import java.util.Collection; import java.util.Map; import org.springframework.beans.factory.BeanFactory; +import org.springframework.context.Lifecycle; import org.springframework.integration.util.AbstractExpressionEvaluator; import org.springframework.integration.util.MessagingMethodInvokerHelper; import org.springframework.messaging.Message; @@ -33,7 +34,8 @@ import org.springframework.messaging.Message; * @author Artem Bilan * @since 2.0 */ -public class MethodInvokingMessageListProcessor extends AbstractExpressionEvaluator { +public class MethodInvokingMessageListProcessor extends AbstractExpressionEvaluator + implements Lifecycle { private final MessagingMethodInvokerHelper delegate; @@ -64,6 +66,17 @@ public class MethodInvokingMessageListProcessor extends AbstractExpressionEva this.delegate.setBeanFactory(beanFactory); } + /** + * A {@code boolean} flag to use SpEL Expression evaluation or + * {@link org.springframework.messaging.handler.invocation.InvocableHandlerMethod} + * for target method invocation. + * @param useSpelInvoker to use SpEL Expression evaluation or not. + * @since 5.0 + */ + public void setUseSpelInvoker(boolean useSpelInvoker) { + this.delegate.setUseSpelInvoker(useSpelInvoker); + } + public String toString() { return this.delegate.toString(); } @@ -80,4 +93,19 @@ public class MethodInvokingMessageListProcessor extends AbstractExpressionEva } } + @Override + public void start() { + this.delegate.start(); + } + + @Override + public void stop() { + this.delegate.stop(); + } + + @Override + public boolean isRunning() { + return this.delegate.isRunning(); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingReleaseStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingReleaseStrategy.java index daaa9a6d28..c11e19068d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingReleaseStrategy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingReleaseStrategy.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -20,6 +20,7 @@ import java.lang.reflect.Method; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.context.Lifecycle; import org.springframework.core.convert.ConversionService; import org.springframework.integration.store.MessageGroup; @@ -28,8 +29,9 @@ import org.springframework.integration.store.MessageGroup; * * @author Marius Bogoevici * @author Dave Syer + * @author Artme Bilan */ -public class MethodInvokingReleaseStrategy implements ReleaseStrategy, BeanFactoryAware { +public class MethodInvokingReleaseStrategy implements ReleaseStrategy, BeanFactoryAware, Lifecycle { private final MethodInvokingMessageListProcessor adapter; @@ -57,4 +59,19 @@ public class MethodInvokingReleaseStrategy implements ReleaseStrategy, BeanFacto return this.adapter.process(messages.getMessages(), null); } + @Override + public void start() { + this.adapter.start(); + } + + @Override + public void stop() { + this.adapter.stop(); + } + + @Override + public boolean isRunning() { + return this.adapter.isRunning(); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java index 6377c82d4f..6807985b0d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,7 +21,6 @@ import java.util.List; import org.aopalliance.aop.Advice; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.reactivestreams.Subscriber; import org.springframework.aop.framework.Advised; import org.springframework.aop.framework.ProxyFactory; @@ -44,7 +43,6 @@ import org.springframework.integration.endpoint.ReactiveConsumer; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.handler.advice.HandleMessageAdvice; import org.springframework.integration.scheduling.PollerMetadata; -import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.PollableChannel; @@ -288,12 +286,7 @@ public class ConsumerEndpointFactoryBean this.endpoint = pollingConsumer; } else { - if (this.handler instanceof Subscriber) { - this.endpoint = new ReactiveConsumer(channel, (Subscriber>) this.handler); - } - else { - this.endpoint = new ReactiveConsumer(channel, this.handler::handleMessage); - } + this.endpoint = new ReactiveConsumer(channel, this.handler); } this.endpoint.setBeanName(this.beanName); this.endpoint.setBeanFactory(this.beanFactory); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java index b34a4d0619..a9175f2420 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -27,7 +27,6 @@ import org.aopalliance.aop.Advice; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.reactivestreams.Publisher; -import org.reactivestreams.Subscriber; import org.springframework.aop.TargetSource; import org.springframework.aop.framework.Advised; @@ -65,7 +64,6 @@ import org.springframework.integration.router.AbstractMessageRouter; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.support.channel.BeanFactoryChannelResolver; import org.springframework.integration.util.MessagingAnnotationUtils; -import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.PollableChannel; @@ -81,8 +79,6 @@ import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; -import reactor.core.publisher.DirectProcessor; - /** * Base class for Method-level annotation post-processors. * @@ -321,17 +317,7 @@ public abstract class AbstractMethodAnnotationPostProcessor> subscriber; - if (handler instanceof Subscriber) { - subscriber = (Subscriber>) handler; - } - else { - //TODO errorConsumer, completeConsumer - DirectProcessor> directProcessor = DirectProcessor.create(); - directProcessor.doOnNext(handler::handleMessage); - subscriber = directProcessor; - } - endpoint = new ReactiveConsumer(inputChannel, subscriber); + endpoint = new ReactiveConsumer(inputChannel, handler); } else { endpoint = new EventDrivenConsumer((SubscribableChannel) inputChannel, handler); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java index fab4580f1d..32cecbb998 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,7 +17,6 @@ package org.springframework.integration.endpoint; import org.springframework.context.Lifecycle; -import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.router.MessageRouter; import org.springframework.integration.support.context.NamedComponent; @@ -95,7 +94,7 @@ public class EventDrivenConsumer extends AbstractEndpoint implements Integration String channelName = ((NamedComponent) this.inputChannel).getComponentName(); String componentType = ((NamedComponent) this.handler).getComponentType(); componentType = StringUtils.hasText(componentType) ? componentType : ""; - String componentName = ((IntegrationObjectSupport) this).getComponentName(); + String componentName = getComponentName(); componentName = (StringUtils.hasText(componentName) && componentName.contains("#")) ? "" : ":" + componentName; StringBuffer buffer = new StringBuffer(); buffer.append("{" + componentType + componentName + "} as a subscriber to the '" + channelName + "' channel"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java index 429819848a..deb7f835ed 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java @@ -22,11 +22,13 @@ import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; +import org.springframework.context.Lifecycle; import org.springframework.integration.channel.MessageChannelReactiveUtils; import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.support.channel.BeanFactoryChannelResolver; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; import org.springframework.util.ErrorHandler; @@ -45,13 +47,18 @@ public class ReactiveConsumer extends AbstractEndpoint { private final Operators.SubscriberAdapter, Message> subscriber; + private final Lifecycle lifecycleDelegate; + private volatile Publisher> publisher; private ErrorHandler errorHandler; - - public ReactiveConsumer(MessageChannel inputChannel, Consumer> consumer) { - this(inputChannel, new ConsumerSubscriber(consumer)); + @SuppressWarnings("unchecked") + public ReactiveConsumer(MessageChannel inputChannel, MessageHandler messageHandler) { + this(inputChannel, + messageHandler instanceof Subscriber + ? (Subscriber>) messageHandler + : new MessageHandlerSubscriber(messageHandler)); } @SuppressWarnings("unchecked") @@ -76,6 +83,7 @@ public class ReactiveConsumer extends AbstractEndpoint { } }; + this.lifecycleDelegate = subscriber instanceof Lifecycle ? (Lifecycle) subscriber : null; } public void setErrorHandler(ErrorHandler errorHandler) { @@ -93,24 +101,34 @@ public class ReactiveConsumer extends AbstractEndpoint { @Override protected void doStart() { + if (this.lifecycleDelegate != null) { + this.lifecycleDelegate.start(); + } this.publisher.subscribe(this.subscriber); } @Override protected void doStop() { this.subscriber.cancel(); + if (this.lifecycleDelegate != null) { + this.lifecycleDelegate.stop(); + } } - private static final class ConsumerSubscriber implements Subscriber>, Receiver, Disposable, Trackable { + private static final class MessageHandlerSubscriber + implements Subscriber>, Receiver, Disposable, Trackable, Lifecycle { private final Consumer> consumer; private Subscription subscription; - ConsumerSubscriber(Consumer> consumer) { - Assert.notNull(consumer); - this.consumer = consumer; + private MessageHandler messageHandler; + + MessageHandlerSubscriber(MessageHandler messageHandler) { + Assert.notNull(messageHandler, "'messageHandler' must not be null"); + this.messageHandler = messageHandler; + this.consumer = this.messageHandler::handleMessage; } @Override @@ -169,6 +187,26 @@ public class ReactiveConsumer extends AbstractEndpoint { return false; } + + @Override + public void start() { + if (this.messageHandler instanceof Lifecycle) { + ((Lifecycle) this.messageHandler).start(); + } + } + + @Override + public void stop() { + if (this.messageHandler instanceof Lifecycle) { + ((Lifecycle) this.messageHandler).stop(); + } + } + + @Override + public boolean isRunning() { + return !(this.messageHandler instanceof Lifecycle) || ((Lifecycle) this.messageHandler).isRunning(); + } + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/MethodInvokingMessageProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/MethodInvokingMessageProcessor.java index e01feb30c0..62900c8a78 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/MethodInvokingMessageProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/MethodInvokingMessageProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -72,6 +72,17 @@ public class MethodInvokingMessageProcessor extends AbstractMessageProcessor< this.delegate.setBeanFactory(beanFactory); } + /** + * A {@code boolean} flag to use SpEL Expression evaluation or + * {@link org.springframework.messaging.handler.invocation.InvocableHandlerMethod} + * for target method invocation. + * @param useSpelInvoker to use SpEL Expression evaluation or not. + * @since 5.0 + */ + public void setUseSpelInvoker(boolean useSpelInvoker) { + this.delegate.setUseSpelInvoker(useSpelInvoker); + } + @Override public void start() { this.delegate.start(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/support/CollectionArgumentResolver.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/support/CollectionArgumentResolver.java new file mode 100644 index 0000000000..fdf66d5b0b --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/support/CollectionArgumentResolver.java @@ -0,0 +1,104 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.handler.support; + +import java.util.Collection; +import java.util.Collections; +import java.util.Iterator; +import java.util.stream.Collectors; + +import org.springframework.core.MethodParameter; +import org.springframework.core.convert.TypeDescriptor; +import org.springframework.integration.util.AbstractExpressionEvaluator; +import org.springframework.messaging.Message; +import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver; +import org.springframework.util.Assert; + +/** + * A {@link HandlerMethodArgumentResolver} implementation for {@link Collection}, + * {@link Iterator} or {@code array} {@link MethodParameter}. + *

+ * If {@link #canProcessMessageList} is set to {@code true}, only messages + * with a payload of {@code Collection>) are supported. + * Depending on the {@link MethodParameter#getNestedParameterType()} the whole + * {@code Collection>} or just payloads of those messages can be use as an actual argument. + *

+ * If the value isn't compatible with {@link MethodParameter}, + * the {@link org.springframework.core.convert.ConversionService} is used + * to convert the value to the target type. + * + * @author Artem Bilan + * + * @since 5.0 + */ +public class CollectionArgumentResolver extends AbstractExpressionEvaluator + implements HandlerMethodArgumentResolver { + + private final boolean canProcessMessageList; + + public CollectionArgumentResolver(boolean canProcessMessageList) { + this.canProcessMessageList = canProcessMessageList; + } + + @Override + public boolean supportsParameter(MethodParameter parameter) { + Class parameterType = parameter.getParameterType(); + return Collection.class.isAssignableFrom(parameterType) + || Iterator.class.isAssignableFrom(parameterType) + || parameterType.isArray(); + } + + @Override + @SuppressWarnings("unchecked") + public Object resolveArgument(MethodParameter parameter, Message message) throws Exception { + Object value = message.getPayload(); + + if (this.canProcessMessageList) { + Assert.state(value instanceof Collection, + "This Argument Resolver only supports messages with a payload of Collection>"); + Collection> messages = (Collection>) value; + + parameter.increaseNestingLevel(); + if (Message.class.isAssignableFrom(parameter.getNestedParameterType())) { + value = messages; + } + else { + value = messages.stream() + .map(Message::getPayload) + .collect(Collectors.toList()); + } + parameter.decreaseNestingLevel(); + } + + if (Iterator.class.isAssignableFrom(parameter.getParameterType())) { + if (value instanceof Iterable) { + return ((Iterable) value).iterator(); + } + else { + return Collections.singleton(value).iterator(); + } + } + else { + return getEvaluationContext() + .getTypeConverter() + .convertValue(value, + TypeDescriptor.forObject(value), + TypeDescriptor.valueOf(parameter.getParameterType())); + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/support/MapArgumentResolver.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/support/MapArgumentResolver.java new file mode 100644 index 0000000000..5c251c6095 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/support/MapArgumentResolver.java @@ -0,0 +1,89 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.handler.support; + +import java.util.Map; +import java.util.Properties; + +import org.springframework.core.MethodParameter; +import org.springframework.core.convert.ConversionService; +import org.springframework.core.convert.TypeDescriptor; +import org.springframework.integration.util.AbstractExpressionEvaluator; +import org.springframework.messaging.Message; +import org.springframework.messaging.handler.annotation.Headers; +import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver; + +/** + * A {@link HandlerMethodArgumentResolver} implementation to resolve argument + * for the {@link MethodParameter} as a {@link Map} or {@link Properties}. + *

+ * The {@link Message#getHeaders()} is used when {@link MethodParameter} is marked + * with the {@link Headers} annotation or {@link Message#getPayload()} isn't {@link Map} + * or {@link Properties} compatible. + *

+ * If {@link MethodParameter} is of {@link Properties} type and {@link Message#getPayload()} + * is a {@link String} containing {@code =} symbol, the {@link MapArgumentResolver} uses + * {@link ConversionService} trying to convert that {@link String} to the {@link Properties} object. + * + * @author Artem Bilan + * + * @since 5.0 + */ +public class MapArgumentResolver extends AbstractExpressionEvaluator + implements HandlerMethodArgumentResolver { + + private static final TypeDescriptor PROPERTIES_TYPE = TypeDescriptor.valueOf(Properties.class); + + @Override + public boolean supportsParameter(MethodParameter parameter) { + return !parameter.hasParameterAnnotation(Payload.class) + && Map.class.isAssignableFrom(parameter.getParameterType()); + } + + @Override + @SuppressWarnings("unchecked") + public Object resolveArgument(MethodParameter parameter, Message message) throws Exception { + Object payload = message.getPayload(); + if (Properties.class.isAssignableFrom(parameter.getParameterType())) { + Map map = message.getHeaders(); + if (!parameter.hasParameterAnnotation(Headers.class)) { + if (payload instanceof Map) { + map = (Map) payload; + } + else if (payload instanceof String && ((String) payload).contains("=")) { + return getEvaluationContext() + .getTypeConverter() + .convertValue(payload, TypeDescriptor.valueOf(String.class), PROPERTIES_TYPE); + } + } + Properties properties = new Properties(); + properties.putAll(map); + return properties; + } + else { + if (!parameter.hasParameterAnnotation(Headers.class) && payload instanceof Map) { + return payload; + } + else { + return message.getHeaders(); + } + } + + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/support/PayloadExpressionArgumentResolver.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/support/PayloadExpressionArgumentResolver.java new file mode 100644 index 0000000000..64069ea231 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/support/PayloadExpressionArgumentResolver.java @@ -0,0 +1,62 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.handler.support; + +import java.util.HashMap; +import java.util.Map; + +import org.springframework.core.MethodParameter; +import org.springframework.expression.Expression; +import org.springframework.integration.util.AbstractExpressionEvaluator; +import org.springframework.messaging.Message; +import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver; +import org.springframework.util.StringUtils; + +/** + * The {@link HandlerMethodArgumentResolver} for evaluating {@link Payload#expression()} + * as a SpEL expression against {@code message} and converting result to expected parameter type. + * + * @author Artem Bilan + * + * @since 5.0 + * + * @see org.springframework.messaging.handler.annotation.support.PayloadArgumentResolver + */ +public class PayloadExpressionArgumentResolver extends AbstractExpressionEvaluator + implements HandlerMethodArgumentResolver { + + private final Map expressionCache = new HashMap<>(); + + @Override + public boolean supportsParameter(MethodParameter parameter) { + Payload ann = parameter.getParameterAnnotation(Payload.class); + return ann != null && StringUtils.hasText(ann.expression()); + } + + @Override + public Object resolveArgument(MethodParameter parameter, Message message) throws Exception { + Expression expression = this.expressionCache.get(parameter); + if (expression == null) { + Payload ann = parameter.getParameterAnnotation(Payload.class); + expression = EXPRESSION_PARSER.parseExpression(ann.expression()); + this.expressionCache.put(parameter, expression); + } + return evaluateExpression(expression, message.getPayload(), parameter.getParameterType()); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/support/PayloadsArgumentResolver.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/support/PayloadsArgumentResolver.java new file mode 100644 index 0000000000..59d4256cfa --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/support/PayloadsArgumentResolver.java @@ -0,0 +1,93 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.handler.support; + +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import org.springframework.core.MethodParameter; +import org.springframework.core.convert.TypeDescriptor; +import org.springframework.expression.Expression; +import org.springframework.integration.annotation.Payloads; +import org.springframework.integration.util.AbstractExpressionEvaluator; +import org.springframework.messaging.Message; +import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +/** + * The {@link HandlerMethodArgumentResolver} for resolving a {@link Collection} + * of {@code payloads} or expression against each {@code payload}. + *

+ * IMPORTANT: The {@link Message} for argument resolution must contain a {@code payload} + * as {@link Collection} of {@link Message}s. + * + * @author Artem Bilan + * + * @since 5.0 + */ +public class PayloadsArgumentResolver extends AbstractExpressionEvaluator + implements HandlerMethodArgumentResolver { + + private final Map expressionCache = new HashMap<>(); + + @Override + public boolean supportsParameter(MethodParameter parameter) { + return parameter.hasParameterAnnotation(Payloads.class); + } + + @Override + @SuppressWarnings("unchecked") + public Object resolveArgument(MethodParameter parameter, Message message) throws Exception { + Object payload = message.getPayload(); + Assert.state(payload instanceof Collection, + "This Argument Resolver support only messages with payload as Collection>"); + Collection> messages = (Collection>) payload; + + if (!this.expressionCache.containsKey(parameter)) { + Payloads payloads = parameter.getParameterAnnotation(Payloads.class); + String expression = payloads.value(); + if (StringUtils.hasText(expression)) { + this.expressionCache.put(parameter, EXPRESSION_PARSER.parseExpression("![payload." + expression + "]")); + } + else { + this.expressionCache.put(parameter, null); + } + + } + + Expression expression = this.expressionCache.get(parameter); + if (expression != null) { + return evaluateExpression(expression, messages, parameter.getParameterType()); + } + else { + List payloads = messages.stream() + .map(Message::getPayload) + .collect(Collectors.toList()); + + return getEvaluationContext() + .getTypeConverter() + .convertValue(payloads, + TypeDescriptor.forObject(payloads), + TypeDescriptor.valueOf(parameter.getParameterType())); + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/support/package-info.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/support/package-info.java new file mode 100644 index 0000000000..7deb8ab5c7 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/support/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides classes for message handlers support. + */ +package org.springframework.integration.handler.support; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessage.java b/spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessage.java index b745b6f55a..27a90b8a7e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessage.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessage.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2015 the original author or authors. + * Copyright 2014-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,7 +21,6 @@ import java.util.Map; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.support.GenericMessage; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; @@ -58,11 +57,6 @@ public class MutableMessage implements Message, Serializable { this.payload = payload; this.headers = new MutableMessageHeaders(headers); - - if (headers != null) { - this.headers.put(MessageHeaders.ID, headers.get(MessageHeaders.ID)); - this.headers.put(MessageHeaders.TIMESTAMP, headers.get(MessageHeaders.TIMESTAMP)); - } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessageHeaders.java b/spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessageHeaders.java index 982eb1ada8..08691735e1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessageHeaders.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessageHeaders.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,6 +17,7 @@ package org.springframework.integration.support; import java.util.Map; +import java.util.UUID; import org.springframework.messaging.MessageHeaders; @@ -27,6 +28,8 @@ import org.springframework.messaging.MessageHeaders; * * @author Stuart Williams * @author David Turanski + * @author Artem Bilan + * * @since 4.2 */ public class MutableMessageHeaders extends MessageHeaders { @@ -34,7 +37,13 @@ public class MutableMessageHeaders extends MessageHeaders { private static final long serialVersionUID = 3084692953798643018L; public MutableMessageHeaders(Map headers) { - super(headers); + super(headers, + (headers != null ? + (UUID) headers.get(MessageHeaders.ID) + : null), + (headers != null ? + (Long) headers.get(MessageHeaders.TIMESTAMP) + : null)); } @Override diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/MessagingMethodInvokerHelper.java b/spring-integration-core/src/main/java/org/springframework/integration/util/MessagingMethodInvokerHelper.java index 8b9cb01cba..b78abbf511 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/MessagingMethodInvokerHelper.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/MessagingMethodInvokerHelper.java @@ -34,33 +34,45 @@ import java.util.Map; import java.util.Properties; import java.util.Set; import java.util.concurrent.atomic.AtomicReference; -import java.util.stream.Stream; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.aop.framework.Advised; import org.springframework.aop.support.AopUtils; +import org.springframework.beans.factory.BeanFactory; import org.springframework.context.Lifecycle; import org.springframework.core.LocalVariableTableParameterNameDiscoverer; import org.springframework.core.MethodParameter; import org.springframework.core.ParameterNameDiscoverer; import org.springframework.core.annotation.AnnotationAttributes; import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.core.convert.ConversionFailedException; +import org.springframework.core.convert.ConversionService; +import org.springframework.core.convert.ConverterNotFoundException; import org.springframework.core.convert.TypeDescriptor; import org.springframework.expression.EvaluationException; import org.springframework.expression.Expression; import org.springframework.expression.TypeConverter; import org.springframework.expression.spel.standard.SpelExpressionParser; -import org.springframework.expression.spel.support.ReflectiveMethodResolver; import org.springframework.expression.spel.support.StandardEvaluationContext; import org.springframework.integration.annotation.Payloads; import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.handler.support.CollectionArgumentResolver; +import org.springframework.integration.handler.support.MapArgumentResolver; +import org.springframework.integration.handler.support.PayloadExpressionArgumentResolver; +import org.springframework.integration.handler.support.PayloadsArgumentResolver; +import org.springframework.integration.support.MutableMessage; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.converter.MessageConversionException; import org.springframework.messaging.handler.annotation.Header; import org.springframework.messaging.handler.annotation.Headers; import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory; +import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver; +import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; +import org.springframework.messaging.handler.invocation.MethodArgumentResolutionException; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; @@ -95,6 +107,9 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator private static final Log logger = LogFactory.getLog(MessagingMethodInvokerHelper.class); + private final DefaultMessageHandlerMethodFactory messageHandlerMethodFactory = + new DefaultMessageHandlerMethodFactory(); + private final Object targetObject; private volatile String displayString; @@ -109,7 +124,7 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator private final HandlerMethod handlerMethod; - private final Class expectedType; + private final TypeDescriptor expectedType; private final boolean canProcessMessageList; @@ -121,6 +136,9 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator private Method method; + private boolean useSpelInvoker; + + public MessagingMethodInvokerHelper(Object targetObject, Method method, Class expectedType, boolean canProcessMessageList) { this(targetObject, null, method, expectedType, canProcessMessageList); @@ -149,13 +167,34 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator this(targetObject, annotationType, (String) null, expectedType, canProcessMessageList); } + /** + * A {@code boolean} flag to use SpEL Expression evaluation or {@link InvocableHandlerMethod} + * for target method invocation. + * @param useSpelInvoker to use SpEL Expression evaluation or not. + * @since 5.0 + */ + public void setUseSpelInvoker(boolean useSpelInvoker) { + this.useSpelInvoker = useSpelInvoker; + } + + @Override + public void setBeanFactory(BeanFactory beanFactory) { + super.setBeanFactory(beanFactory); + this.messageHandlerMethodFactory.setBeanFactory(beanFactory); + } + + @Override + public void setConversionService(ConversionService conversionService) { + super.setConversionService(conversionService); + this.messageHandlerMethodFactory.setConversionService(conversionService); + } public T process(Message message) throws Exception { ParametersWrapper parameters = new ParametersWrapper(message); return processInternal(parameters); } - public T process(Collection> messages, Map headers) throws Exception { + public T process(Collection> messages, Map headers) throws Exception { ParametersWrapper parameters = new ParametersWrapper(messages, headers); return processInternal(parameters); } @@ -194,16 +233,22 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator this.canProcessMessageList = canProcessMessageList; Assert.notNull(method, "method must not be null"); this.method = method; - this.expectedType = expectedType; this.requiresReply = expectedType != null; if (expectedType != null) { Assert.isTrue(method.getReturnType() != Void.class && method.getReturnType() != Void.TYPE, "method must have a return type"); + this.expectedType = TypeDescriptor.valueOf(expectedType); } + else { + this.expectedType = null; + } + Assert.notNull(targetObject, "targetObject must not be null"); this.targetObject = targetObject; try { - this.handlerMethod = new HandlerMethod(method, canProcessMessageList); + InvocableHandlerMethod invocableHandlerMethod = + this.messageHandlerMethodFactory.createInvocableHandlerMethod(targetObject, method); + this.handlerMethod = new HandlerMethod(invocableHandlerMethod, canProcessMessageList); } catch (IneligibleMethodException e) { throw new IllegalArgumentException(e); @@ -220,11 +265,15 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator this.methodName = methodName; this.canProcessMessageList = canProcessMessageList; Assert.notNull(targetObject, "targetObject must not be null"); - this.expectedType = expectedType; + if (expectedType != null) { + this.expectedType = TypeDescriptor.valueOf(expectedType); + } + else { + this.expectedType = null; + } this.targetObject = targetObject; - this.requiresReply = expectedType != null; Map, HandlerMethod>> handlerMethodsForTarget = - findHandlerMethodsForTarget(targetObject, annotationType, methodName, this.requiresReply); + findHandlerMethodsForTarget(targetObject, annotationType, methodName, expectedType != null); Map, HandlerMethod> handlerMethods = handlerMethodsForTarget.get(CANDIDATE_METHODS); Map, HandlerMethod> handlerMessageMethods = handlerMethodsForTarget.get(CANDIDATE_MESSAGE_METHODS); if ((handlerMethods.size() == 1 && handlerMessageMethods.isEmpty()) || @@ -268,39 +317,22 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator private void prepareEvaluationContext() { StandardEvaluationContext context = getEvaluationContext(false); Class targetType = AopUtils.getTargetClass(this.targetObject); - - ReflectiveMethodResolver declaredMethodResolver = new ReflectiveMethodResolver() { - - @Override - protected Method[] getMethods(Class type) { - return Stream.of(type.getMethods(), type.getDeclaredMethods()) - .flatMap(Stream::of) - .toArray(Method[]::new); - } - - }; - if (this.method != null) { - FixedMethodFilter fixedMethodFilter = new FixedMethodFilter(this.method); - context.registerMethodFilter(targetType, fixedMethodFilter); - declaredMethodResolver.registerMethodFilter(targetType, fixedMethodFilter); + context.registerMethodFilter(targetType, new FixedMethodFilter(this.method)); if (this.expectedType != null) { Assert.state(context.getTypeConverter() - .canConvert(TypeDescriptor.valueOf((this.method).getReturnType()), - TypeDescriptor.valueOf(this.expectedType)), + .canConvert(TypeDescriptor.valueOf((this.method).getReturnType()), this.expectedType), "Cannot convert to expected type (" + this.expectedType + ") from " + this.method); } } else { - AnnotatedMethodFilter annotatedMethodFilter = new AnnotatedMethodFilter(this.annotationType, - this.methodName, this.requiresReply); - Assert.state(canReturnExpectedType(annotatedMethodFilter, targetType, context.getTypeConverter()), + AnnotatedMethodFilter filter = new AnnotatedMethodFilter(this.annotationType, this.methodName, + this.requiresReply); + Assert.state(canReturnExpectedType(filter, targetType, context.getTypeConverter()), "Cannot convert to expected type (" + this.expectedType + ") from " + this.method); - context.registerMethodFilter(targetType, annotatedMethodFilter); - declaredMethodResolver.registerMethodFilter(targetType, annotatedMethodFilter); + context.registerMethodFilter(targetType, filter); } context.setVariable("target", this.targetObject); - context.setMethodResolvers(Collections.singletonList(declaredMethodResolver)); } private boolean canReturnExpectedType(AnnotatedMethodFilter filter, Class targetType, @@ -310,39 +342,100 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator } List methods = filter.filter(Arrays.asList(ReflectionUtils.getAllDeclaredMethods(targetType))); for (Method method : methods) { - if (typeConverter.canConvert(TypeDescriptor.valueOf(method.getReturnType()), - TypeDescriptor.valueOf(this.expectedType))) { + if (typeConverter.canConvert(TypeDescriptor.valueOf(method.getReturnType()), this.expectedType)) { return true; } } return false; } + @SuppressWarnings("unchecked") private T processInternal(ParametersWrapper parameters) throws Exception { if (!this.initialized) { synchronized (this) { if (!this.initialized) { + PayloadExpressionArgumentResolver payloadExpressionArgumentResolver = + new PayloadExpressionArgumentResolver(); + payloadExpressionArgumentResolver.setBeanFactory(getBeanFactory()); + + PayloadsArgumentResolver payloadsArgumentResolver = new PayloadsArgumentResolver(); + payloadsArgumentResolver.setBeanFactory(getBeanFactory()); + + CollectionArgumentResolver collectionArgumentResolver = + new CollectionArgumentResolver(this.canProcessMessageList); + collectionArgumentResolver.setBeanFactory(getBeanFactory()); + + MapArgumentResolver mapArgumentResolver = new MapArgumentResolver(); + mapArgumentResolver.setBeanFactory(getBeanFactory()); + + List customArgumentResolvers = new LinkedList<>(); + customArgumentResolvers.add(payloadExpressionArgumentResolver); + customArgumentResolvers.add(payloadsArgumentResolver); + customArgumentResolvers.add(collectionArgumentResolver); + customArgumentResolvers.add(mapArgumentResolver); + + this.messageHandlerMethodFactory.setCustomArgumentResolvers(customArgumentResolvers); + this.messageHandlerMethodFactory.afterPropertiesSet(); prepareEvaluationContext(); this.initialized = true; } } } HandlerMethod candidate = this.findHandlerMethodForParameters(parameters); + Expression expression = candidate.expression; Assert.notNull(candidate, "No candidate methods found for messages."); - Expression expression = candidate.getExpression(); - Class expectedType = this.expectedType != null ? this.expectedType : candidate.method.getReturnType(); + + T result = null; try { - @SuppressWarnings("unchecked") - T result = (T) evaluateExpression(expression, parameters, expectedType); - if (this.requiresReply) { - Assert.notNull(result, - "Expression evaluation result was null, but this processor requires a reply."); + if (this.useSpelInvoker || candidate.spelOnly) { + result = invokeExpression(expression, parameters); } + else { + result = candidate.invoke(parameters); + } + } + catch (MethodArgumentResolutionException | MessageConversionException | IllegalStateException e) { + if (e instanceof MessageConversionException) { + if (e.getCause() instanceof ConversionFailedException && + !(e.getCause().getCause() instanceof ConverterNotFoundException)) { + throw e; + } + } + else if (e instanceof IllegalStateException) { + if (e.getCause() instanceof IllegalArgumentException + && !"argument type mismatch".equals(e.getCause().getMessage())) { + throw e; + } + } + if (logger.isInfoEnabled()) { + logger.info("Failed to invoke [ " + candidate.invocableHandlerMethod + + "] with provided arguments [ " + parameters + " ]. \n" + + "Falling back to SpEL invocation for expression [ " + + expression.getExpressionString() + " ]"); + } + + result = invokeExpression(expression, parameters); + } + + if (result != null && this.expectedType != null) { + return (T) getEvaluationContext(true) + .getTypeConverter() + .convertValue(result, TypeDescriptor.forObject(result), this.expectedType); + } + else { return result; } + } + + @SuppressWarnings("unchecked") + private T invokeExpression(Expression expression, ParametersWrapper parameters) throws Exception { + try { + return (T) evaluateExpression(expression, parameters); + } catch (Exception e) { Throwable evaluationException = e; - if ((e instanceof EvaluationException || e instanceof MessageHandlingException) && e.getCause() != null) { + if ((e instanceof EvaluationException || e instanceof MessageHandlingException) + && e.getCause() != null) { evaluationException = e.getCause(); } if (evaluationException instanceof Exception) { @@ -357,14 +450,14 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator private Map, HandlerMethod>> findHandlerMethodsForTarget(final Object targetObject, final Class annotationType, final String methodName, final boolean requiresReply) { - Map, HandlerMethod>> handlerMethods = new HashMap, HandlerMethod>>(); + Map, HandlerMethod>> handlerMethods = new HashMap<>(); - final Map, HandlerMethod> candidateMethods = new HashMap, HandlerMethod>(); - final Map, HandlerMethod> candidateMessageMethods = new HashMap, HandlerMethod>(); - final Map, HandlerMethod> fallbackMethods = new HashMap, HandlerMethod>(); - final Map, HandlerMethod> fallbackMessageMethods = new HashMap, HandlerMethod>(); - final AtomicReference> ambiguousFallbackType = new AtomicReference>(); - final AtomicReference> ambiguousFallbackMessageGenericType = new AtomicReference>(); + final Map, HandlerMethod> candidateMethods = new HashMap<>(); + final Map, HandlerMethod> candidateMessageMethods = new HashMap<>(); + final Map, HandlerMethod> fallbackMethods = new HashMap<>(); + final Map, HandlerMethod> fallbackMessageMethods = new HashMap<>(); + final AtomicReference> ambiguousFallbackType = new AtomicReference<>(); + final AtomicReference> ambiguousFallbackMessageGenericType = new AtomicReference<>(); final Class targetClass = getTargetClass(targetObject); MethodFilter methodFilter = new UniqueMethodFilter(targetClass); ReflectionUtils.doWithMethods(targetClass, method1 -> { @@ -394,9 +487,11 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator && ObjectUtils.containsElement(new String[] { "start", "stop", "isRunning" }, method1.getName())) { return; } - HandlerMethod handlerMethod1 = null; + HandlerMethod handlerMethod1; try { - handlerMethod1 = new HandlerMethod(method1, this.canProcessMessageList); + InvocableHandlerMethod invocableHandlerMethod = + this.messageHandlerMethodFactory.createInvocableHandlerMethod(targetObject, method1); + handlerMethod1 = new HandlerMethod(invocableHandlerMethod, this.canProcessMessageList); } catch (IneligibleMethodException e) { if (logger.isDebugEnabled()) { @@ -476,7 +571,7 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator * Ambiguous means > 1 method that takes the same payload type, or > 1 method * that takes a Message with the same generic type. */ - List frameworkMethods = new ArrayList(); + List frameworkMethods = new ArrayList<>(); Class[] allInterfaces = org.springframework.util.ClassUtils.getAllInterfacesForClass(targetClass); for (Class iface : allInterfaces) { try { @@ -493,8 +588,11 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator } } if (frameworkMethods.size() == 1) { - HandlerMethod handlerMethod = new HandlerMethod(frameworkMethods.get(0), this.canProcessMessageList); - handlerMethods.put(CANDIDATE_METHODS, Collections., HandlerMethod>singletonMap(Object.class, handlerMethod)); + InvocableHandlerMethod invocableHandlerMethod = + this.messageHandlerMethodFactory.createInvocableHandlerMethod(targetObject, + frameworkMethods.get(0)); + HandlerMethod handlerMethod = new HandlerMethod(invocableHandlerMethod, this.canProcessMessageList); + handlerMethods.put(CANDIDATE_METHODS, Collections.singletonMap(Object.class, handlerMethod)); handlerMethods.put(CANDIDATE_MESSAGE_METHODS, candidateMessageMethods); return handlerMethods; } @@ -507,7 +605,9 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator Assert.isNull(ambiguousFallbackType.get(), "Found ambiguous parameter type [" + ambiguousFallbackType + "] for method match: " + fallbackMethods.values()); Assert.isNull(ambiguousFallbackMessageGenericType.get(), - "Found ambiguous parameter type [" + ambiguousFallbackMessageGenericType + "] for method match: " + "Found ambiguous parameter type [" + + ambiguousFallbackMessageGenericType + + "] for method match: " + fallbackMethods.values()); handlerMethods.put(CANDIDATE_METHODS, fallbackMethods); @@ -519,7 +619,7 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator Map, HandlerMethod> candidateMessageMethods, Map, HandlerMethod> candidateMethods) { if (AopUtils.isAopProxy(targetObject)) { - final AtomicReference targetMethod = new AtomicReference(); + final AtomicReference targetMethod = new AtomicReference<>(); Class[] interfaces = ((Advised) targetObject).getProxiedInterfaces(); for (Class clazz : interfaces) { ReflectionUtils.doWithMethods(clazz, method1 -> { @@ -533,7 +633,9 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator } Method method = targetMethod.get(); if (method != null) { - HandlerMethod handlerMethod = new HandlerMethod(method, this.canProcessMessageList); + InvocableHandlerMethod invocableHandlerMethod = + this.messageHandlerMethodFactory.createInvocableHandlerMethod(targetObject, method); + HandlerMethod handlerMethod = new HandlerMethod(invocableHandlerMethod, this.canProcessMessageList); Class targetParameterType = handlerMethod.getTargetParameterType(); if (handlerMethod.isMessageMethod()) { if (candidateMessageMethods.containsKey(targetParameterType)) { @@ -653,11 +755,10 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator @SuppressWarnings("unused") private static final Collection> dummyMessages = Collections.emptyList(); - - private final Method method; - private final Expression expression; + private final InvocableHandlerMethod invocableHandlerMethod; + private final boolean canProcessMessageList; private volatile TypeDescriptor targetParameterTypeDescriptor; @@ -666,14 +767,25 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator private volatile boolean messageMethod; - HandlerMethod(Method method, boolean canProcessMessageList) { - this.method = method; + private volatile boolean spelOnly; + + HandlerMethod(InvocableHandlerMethod invocableHandlerMethod, boolean canProcessMessageList) { + this.invocableHandlerMethod = invocableHandlerMethod; this.canProcessMessageList = canProcessMessageList; - this.expression = this.generateExpression(method); + this.expression = generateExpression(this.invocableHandlerMethod.getMethod()); } - Expression getExpression() { + @SuppressWarnings("unchecked") + public T invoke(ParametersWrapper parameters) throws Exception { + Message message = parameters.getMessage(); + if (this.canProcessMessageList) { + message = new MutableMessage<>(parameters.getMessages(), parameters.getHeaders()); + } + return (T) this.invocableHandlerMethod.invoke(message); + } + + public Expression getExpression() { return this.expression; } @@ -687,7 +799,7 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator @Override public String toString() { - return this.method.toString(); + return this.invocableHandlerMethod.toString(); } private Expression generateExpression(Method method) { @@ -718,6 +830,10 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator } } if (annotationType.equals(Payloads.class)) { + Assert.isTrue(this.canProcessMessageList, + "The @Payloads annotation can only be applied if method handler canProcessMessageList."); + Assert.isTrue(Collection.class.isAssignableFrom(parameterType), + "The @Payloads annotation can only be applied to a Collection-typed parameter."); sb.append("messages.![payload"); String qualifierExpression = ((Payloads) mappingAnnotation).value(); if (StringUtils.hasText(qualifierExpression)) { @@ -827,6 +943,7 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator headerName = tokens[0]; if (StringUtils.hasText(tokens[1])) { relativeExpression = "." + tokens[1]; + this.spelOnly = true; } } else { @@ -844,7 +961,7 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator return headerRetrievalExpression + " != null ? " + fullHeaderExpression + " : " + fallbackExpression; } - private synchronized void setExclusiveTargetParameterType(TypeDescriptor targetParameterType, + private void setExclusiveTargetParameterType(TypeDescriptor targetParameterType, MethodParameter methodParameter) { if (this.targetParameterTypeDescriptor != null) { throw new IneligibleMethodException("Found more than one parameter type candidate: [" + @@ -862,24 +979,24 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator } } - public class ParametersWrapper { + private static class ParametersWrapper { private final Object payload; private final Collection> messages; - private final Map headers; + private final Map headers; private final Message message; - public ParametersWrapper(Message message) { + ParametersWrapper(Message message) { this.message = message; this.payload = message.getPayload(); this.headers = message.getHeaders(); this.messages = null; } - public ParametersWrapper(Collection> messages, Map headers) { + ParametersWrapper(Collection> messages, Map headers) { this.payload = null; this.messages = messages; this.headers = headers; @@ -887,16 +1004,18 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator } public Object getPayload() { - Assert.state(this.payload != null, "Invalid method parameter for payload: was expecting collection."); + Assert.state(this.payload != null, + "Invalid method parameter for payload: was expecting collection."); return this.payload; } public Collection> getMessages() { - Assert.state(this.messages != null, "Invalid method parameter for messages: was expecting a single payload."); + Assert.state(this.messages != null, + "Invalid method parameter for messages: was expecting a single payload."); return this.messages; } - public Map getHeaders() { + public Map getHeaders() { return this.headers; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java index 89610927f5..471f415d93 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -20,6 +20,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; import java.util.Arrays; import java.util.Collections; @@ -27,9 +28,11 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import org.junit.Before; import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.support.MessageBuilder; @@ -49,6 +52,13 @@ public class AggregatingMessageGroupProcessorHeaderTests { private final MethodInvokingMessageGroupProcessor methodInvokingProcessor = new MethodInvokingMessageGroupProcessor(new TestAggregatorBean(), "aggregate"); + @Before + public void setup() { + this.defaultProcessor.setBeanFactory(mock(BeanFactory.class)); + this.methodInvokingProcessor.setBeanFactory(mock(BeanFactory.class)); + } + + @Test public void singleMessageUsingDefaultProcessor() { this.singleMessage(defaultProcessor); @@ -283,6 +293,7 @@ public class AggregatingMessageGroupProcessorHeaderTests { } return sb.toString(); } + } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java index 93a5b82718..271c006722 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java @@ -53,6 +53,15 @@ import org.springframework.messaging.handler.annotation.Header; import org.springframework.messaging.handler.annotation.Headers; import org.springframework.messaging.support.GenericMessage; +/** + * @author Iwein Fuld + * @author Dave Syer + * @author Mark Fisher + * @author Oleg Zhurakousky + * @author Gunnar Hillert + * @author Gary Russell + * @author Artem Bilan + */ @RunWith(MockitoJUnitRunner.class) public class MethodInvokingMessageGroupProcessorTests { @@ -93,7 +102,7 @@ public class MethodInvokingMessageGroupProcessorTests { MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new AnnotatedAggregatorMethod()); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertThat((Integer) ((Message) result).getPayload(), is(7)); + assertThat(((Message) result).getPayload(), is(7)); } @Test @@ -101,6 +110,7 @@ public class MethodInvokingMessageGroupProcessorTests { @SuppressWarnings("unused") class SimpleAggregator { + public Integer and(List flags) { int result = 0; for (Integer flag : flags) { @@ -113,7 +123,7 @@ public class MethodInvokingMessageGroupProcessorTests { MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator()); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertThat((Integer) ((Message) result).getPayload(), is(7)); + assertThat(((Message) result).getPayload(), is(7)); } @Test @@ -121,6 +131,7 @@ public class MethodInvokingMessageGroupProcessorTests { @SuppressWarnings("unused") class SimpleAggregator { + public Integer and(List> flags) { int result = 0; for (Message flag : flags) { @@ -133,7 +144,7 @@ public class MethodInvokingMessageGroupProcessorTests { MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator()); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertThat((Integer) ((Message) result).getPayload(), is(7)); + assertThat(((Message) result).getPayload(), is(7)); } @Test @@ -141,6 +152,7 @@ public class MethodInvokingMessageGroupProcessorTests { @SuppressWarnings("unused") class SimpleAggregator { + public String and(List flags, @Header("foo") List header) { List result = new ArrayList(); for (int flag : flags) { @@ -157,7 +169,7 @@ public class MethodInvokingMessageGroupProcessorTests { messagesUpForProcessing.add(MessageBuilder.withPayload(3).setHeader("foo", Arrays.asList(101, 102)).build()); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertThat((String) ((Message) result).getPayload(), is("[1, 2, 4, 3, 101, 102]")); + assertThat(((Message) result).getPayload(), is("[1, 2, 4, 3, 101, 102]")); } @Test @@ -165,6 +177,7 @@ public class MethodInvokingMessageGroupProcessorTests { @SuppressWarnings("unused") class SimpleAggregator { + public String and(@Payloads List rawFlags, @Header("foo") List header) { @SuppressWarnings("unchecked") List flags = (List) rawFlags; @@ -183,7 +196,7 @@ public class MethodInvokingMessageGroupProcessorTests { messagesUpForProcessing.add(MessageBuilder.withPayload(3).setHeader("foo", Arrays.asList(101, 102)).build()); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertThat((String) ((Message) result).getPayload(), is("[1, 2, 4, 3, 101, 102]")); + assertThat(((Message) result).getPayload(), is("[1, 2, 4, 3, 101, 102]")); } @Test @@ -191,6 +204,7 @@ public class MethodInvokingMessageGroupProcessorTests { @SuppressWarnings("unused") class SimpleAggregator { + @Aggregator public String and(@Payloads List flags) { List result = new ArrayList(); @@ -199,6 +213,7 @@ public class MethodInvokingMessageGroupProcessorTests { } return result.toString(); } + public String or(List flags) { throw new UnsupportedOperationException("Not expected"); } @@ -207,7 +222,7 @@ public class MethodInvokingMessageGroupProcessorTests { MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator()); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertThat((String) ((Message) result).getPayload(), is("[1, 2, 4]")); + assertThat(((Message) result).getPayload(), is("[1, 2, 4]")); } @Test @@ -215,6 +230,7 @@ public class MethodInvokingMessageGroupProcessorTests { @SuppressWarnings("unused") class SimpleAggregator { + public Integer and(Collection flags) { int result = 0; for (Integer flag : flags) { @@ -227,7 +243,7 @@ public class MethodInvokingMessageGroupProcessorTests { MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator()); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertThat((Integer) ((Message) result).getPayload(), is(7)); + assertThat(((Message) result).getPayload(), is(7)); } @Test @@ -235,6 +251,7 @@ public class MethodInvokingMessageGroupProcessorTests { @SuppressWarnings("unused") class SimpleAggregator { + public Integer and(int[] flags) { int result = 0; for (int flag : flags) { @@ -247,7 +264,7 @@ public class MethodInvokingMessageGroupProcessorTests { MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator()); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertThat((Integer) ((Message) result).getPayload(), is(7)); + assertThat(((Message) result).getPayload(), is(7)); } @@ -256,6 +273,7 @@ public class MethodInvokingMessageGroupProcessorTests { @SuppressWarnings("unused") class SimpleAggregator { + public Integer and(Iterator flags) { int result = 0; while (flags.hasNext()) { @@ -268,15 +286,17 @@ public class MethodInvokingMessageGroupProcessorTests { MethodInvokingMessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator()); GenericConversionService conversionService = new DefaultConversionService(); conversionService.addConverter(new Converter, Iterator>() { + @Override public Iterator convert(ArrayList source) { return source.iterator(); } + }); processor.setConversionService(conversionService); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertThat((Integer) ((Message) result).getPayload(), is(7)); + assertThat(((Message) result).getPayload(), is(7)); } @Test @@ -284,6 +304,7 @@ public class MethodInvokingMessageGroupProcessorTests { @SuppressWarnings("unused") class UnannotatedAggregator { + public Integer and(List flags) { int result = 0; for (Integer flag : flags) { @@ -305,7 +326,7 @@ public class MethodInvokingMessageGroupProcessorTests { MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new UnannotatedAggregator()); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertThat((Integer) ((Message) result).getPayload(), is(7)); + assertThat(((Message) result).getPayload(), is(7)); } @Test @@ -313,6 +334,7 @@ public class MethodInvokingMessageGroupProcessorTests { @SuppressWarnings("unused") class UnannotatedAggregator { + public Iterator and(Iterator> flags) { return flags; @@ -392,10 +414,12 @@ public class MethodInvokingMessageGroupProcessorTests { public void testHeaderParameters() throws Exception { class SingleAnnotationTestBean { + @Aggregator public String method1(List input, @Header("foo") String foo) { return input.get(0) + foo; } + } SingleAnnotationTestBean bean = new SingleAnnotationTestBean(); @@ -410,10 +434,12 @@ public class MethodInvokingMessageGroupProcessorTests { public void testHeadersParameters() throws Exception { class SingleAnnotationTestBean { + @Aggregator public String method1(List input, @Headers Map map) { return input.get(0) + map.get("foo"); } + } SingleAnnotationTestBean bean = new SingleAnnotationTestBean(); @@ -507,14 +533,16 @@ public class MethodInvokingMessageGroupProcessorTests { QueueChannel output = new QueueChannel(); GreetingService testBean = new GreetingBean(); ProxyFactory proxyFactory = new ProxyFactory(testBean); - proxyFactory.setProxyTargetClass(false); + proxyFactory.setProxyTargetClass(true); testBean = (GreetingService) proxyFactory.getProxy(); MethodInvokingMessageGroupProcessor aggregator = new MethodInvokingMessageGroupProcessor(testBean); AggregatingMessageHandler handler = new AggregatingMessageHandler(aggregator); handler.setReleaseStrategy(new MessageCountReleaseStrategy()); handler.setOutputChannel(output); + EventDrivenConsumer endpoint = new EventDrivenConsumer(input, handler); endpoint.start(); + Message message = MessageBuilder.withPayload("proxy").setCorrelationId("abc").build(); input.send(message); assertEquals("hello proxy", output.receive(0).getPayload()); @@ -534,6 +562,7 @@ public class MethodInvokingMessageGroupProcessorTests { handler.setOutputChannel(output); EventDrivenConsumer endpoint = new EventDrivenConsumer(input, handler); endpoint.start(); + Message message = MessageBuilder.withPayload("proxy").setCorrelationId("abc").build(); input.send(message); assertEquals("hello proxy", output.receive(0).getPayload()); @@ -541,7 +570,9 @@ public class MethodInvokingMessageGroupProcessorTests { public interface GreetingService { + String sayHello(List names); + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java index 99b9573ee2..c83d571bbf 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -49,6 +49,7 @@ import org.springframework.scheduling.support.PeriodicTrigger; /** * @author Mark Fisher + * @author Artem Bilan */ public class ApplicationContextMessageBusTests { @@ -74,7 +75,7 @@ public class ApplicationContextMessageBusTests { endpoint.setBeanFactory(mock(BeanFactory.class)); context.registerEndpoint("testEndpoint", endpoint); context.refresh(); - Message result = targetChannel.receive(3000); + Message result = targetChannel.receive(10000); assertEquals("test", result.getPayload()); context.stop(); } @@ -88,19 +89,19 @@ public class ApplicationContextMessageBusTests { QueueChannel targetChannel = new QueueChannel(); context.registerChannel("targetChannel", targetChannel); context.refresh(); - Message result = targetChannel.receive(100); + Message result = targetChannel.receive(10); assertNull(result); context.stop(); } @Test - public void autodetectionWithApplicationContext() { + public void autoDetectionWithApplicationContext() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("messageBusTests.xml", this.getClass()); context.start(); PollableChannel sourceChannel = (PollableChannel) context.getBean("sourceChannel"); sourceChannel.send(new GenericMessage("test")); PollableChannel targetChannel = (PollableChannel) context.getBean("targetChannel"); - Message result = targetChannel.receive(3000); + Message result = targetChannel.receive(10000); assertEquals("test", result.getPayload()); context.close(); } @@ -136,7 +137,7 @@ public class ApplicationContextMessageBusTests { context.registerEndpoint("testEndpoint2", endpoint2); context.refresh(); inputChannel.send(new GenericMessage("testing")); - Message message1 = outputChannel1.receive(3000); + Message message1 = outputChannel1.receive(10000); Message message2 = outputChannel2.receive(0); context.stop(); assertTrue("exactly one message should be null", message1 == null ^ message2 == null); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml index 89b1ecb18f..1e221d65b3 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml @@ -33,4 +33,6 @@ + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveConsumerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveConsumerTests.java index 2fcea32dfd..2cddf87560 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveConsumerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveConsumerTests.java @@ -75,7 +75,7 @@ public class ReactiveConsumerTests { stopLatch.countDown(); }; - MethodInvokingMessageHandler testSubscriber = new MethodInvokingMessageHandler(messageHandler, (String) null); + MessageHandler testSubscriber = new MethodInvokingMessageHandler(messageHandler, (String) null); ReactiveConsumer reactiveConsumer = new ReactiveConsumer(testChannel, testSubscriber); reactiveConsumer.setBeanFactory(mock(BeanFactory.class)); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java index 769d277fc9..36e5dc147a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -30,6 +30,7 @@ import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; import org.springframework.context.ApplicationContext; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.aggregator.MethodInvokingCorrelationStrategy; import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy; @@ -48,7 +49,7 @@ public class AggregatorAnnotationTests { @Test public void testAnnotationWithDefaultSettings() { - ApplicationContext context = new ClassPathXmlApplicationContext( + ConfigurableApplicationContext context = new ClassPathXmlApplicationContext( new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" }); final String endpointName = "endpointWithDefaultAnnotation"; MessageHandler aggregator = this.getAggregator(context, endpointName); @@ -57,11 +58,12 @@ public class AggregatorAnnotationTests { assertTrue(getPropertyValue(aggregator, "discardChannel") instanceof NullChannel); assertEquals(-1L, getPropertyValue(aggregator, "messagingTemplate.sendTimeout")); assertEquals(false, getPropertyValue(aggregator, "sendPartialResultOnExpiry")); + context.close(); } @Test public void testAnnotationWithCustomSettings() { - ApplicationContext context = new ClassPathXmlApplicationContext( + ConfigurableApplicationContext context = new ClassPathXmlApplicationContext( new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" }); final String endpointName = "endpointWithCustomizedAnnotation"; MessageHandler aggregator = this.getAggregator(context, endpointName); @@ -70,28 +72,30 @@ public class AggregatorAnnotationTests { assertEquals("discardChannel", getPropertyValue(aggregator, "discardChannelName")); assertEquals(98765432L, getPropertyValue(aggregator, "messagingTemplate.sendTimeout")); assertEquals(true, getPropertyValue(aggregator, "sendPartialResultOnExpiry")); + context.close(); } @Test public void testAnnotationWithCustomReleaseStrategy() throws Exception { - ApplicationContext context = new ClassPathXmlApplicationContext( + ConfigurableApplicationContext context = new ClassPathXmlApplicationContext( new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" }); final String endpointName = "endpointWithDefaultAnnotationAndCustomReleaseStrategy"; MessageHandler aggregator = this.getAggregator(context, endpointName); Object releaseStrategy = getPropertyValue(aggregator, "releaseStrategy"); Assert.assertTrue(releaseStrategy instanceof MethodInvokingReleaseStrategy); MethodInvokingReleaseStrategy releaseStrategyAdapter = (MethodInvokingReleaseStrategy) releaseStrategy; - Object handlerMethods = new DirectFieldAccessor(new DirectFieldAccessor(new DirectFieldAccessor(releaseStrategyAdapter) - .getPropertyValue("adapter")).getPropertyValue("delegate")).getPropertyValue("handlerMethods"); + Object handlerMethods = new DirectFieldAccessor(releaseStrategyAdapter) + .getPropertyValue("adapter.delegate.handlerMethods"); assertNull(handlerMethods); - Object handlerMethod = new DirectFieldAccessor(new DirectFieldAccessor(new DirectFieldAccessor(releaseStrategyAdapter) - .getPropertyValue("adapter")).getPropertyValue("delegate")).getPropertyValue("handlerMethod"); + Object handlerMethod = new DirectFieldAccessor(releaseStrategyAdapter) + .getPropertyValue("adapter.delegate.handlerMethod"); assertTrue(handlerMethod.toString().contains("completionChecker")); + context.close(); } @Test public void testAnnotationWithCustomCorrelationStrategy() throws Exception { - ApplicationContext context = new ClassPathXmlApplicationContext( + ConfigurableApplicationContext context = new ClassPathXmlApplicationContext( new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" }); final String endpointName = "endpointWithCorrelationStrategy"; MessageHandler aggregator = this.getAggregator(context, endpointName); @@ -106,8 +110,9 @@ public class AggregatorAnnotationTests { Object handlerMethod = processorAccessor.getPropertyValue("handlerMethod"); assertNotNull(handlerMethod); DirectFieldAccessor handlerMethodAccessor = new DirectFieldAccessor(handlerMethod); - Method completionCheckerMethod = (Method) handlerMethodAccessor.getPropertyValue("method"); + Method completionCheckerMethod = (Method) handlerMethodAccessor.getPropertyValue("invocableHandlerMethod.method"); assertEquals("correlate", completionCheckerMethod.getName()); + context.close(); } private MessageHandler getAggregator(ApplicationContext context, final String endpointName) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/SubscriberOrderTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/SubscriberOrderTests.java index 946ad5f983..8f0b7375d1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/SubscriberOrderTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/SubscriberOrderTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -32,18 +32,20 @@ import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; +import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.support.GenericMessage; /** * @author Mark Fisher + * @author Artem Bilan */ public class SubscriberOrderTests { @Test public void directChannelAndFailoverDispatcherWithSingleCallPerMethod() { - GenericApplicationContext context = new GenericApplicationContext(); + GenericApplicationContext context = TestUtils.createTestApplicationContext(); context.registerBeanDefinition("postProcessor", new RootBeanDefinition(MessagingAnnotationPostProcessor.class)); RootBeanDefinition channelDefinition = new RootBeanDefinition(DirectChannel.class); context.registerBeanDefinition("input", channelDefinition); @@ -70,7 +72,7 @@ public class SubscriberOrderTests { @Test public void directChannelAndFailoverDispatcherWithMultipleCallsPerMethod() { - GenericApplicationContext context = new GenericApplicationContext(); + GenericApplicationContext context = TestUtils.createTestApplicationContext(); context.registerBeanDefinition("postProcessor", new RootBeanDefinition(MessagingAnnotationPostProcessor.class)); BeanDefinitionBuilder channelBuilder = BeanDefinitionBuilder.rootBeanDefinition(DirectChannel.class); channelBuilder.addConstructorArgValue(null); @@ -112,7 +114,7 @@ public class SubscriberOrderTests { @Test public void directChannelAndRoundRobinDispatcher() { - GenericApplicationContext context = new GenericApplicationContext(); + GenericApplicationContext context = TestUtils.createTestApplicationContext(); context.registerBeanDefinition("postProcessor", new RootBeanDefinition(MessagingAnnotationPostProcessor.class)); RootBeanDefinition channelDefinition = new RootBeanDefinition(DirectChannel.class); channelDefinition.getConstructorArgumentValues().addGenericArgumentValue(new RoundRobinLoadBalancingStrategy()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/MessageProducerSupportTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/MessageProducerSupportTests.java index d3be8d1cbc..003d98ce2f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/MessageProducerSupportTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/MessageProducerSupportTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -20,14 +20,12 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; -import static org.mockito.Mockito.mock; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; -import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.handler.ServiceActivatingHandler; @@ -43,6 +41,7 @@ import org.springframework.messaging.support.GenericMessage; * @author Mark Fisher * @author Gary Russell * @author Kris Jacyna + * @author Artem Bilan * @since 2.0.1 */ public class MessageProducerSupportTests { @@ -83,6 +82,8 @@ public class MessageProducerSupportTests { @Test public void validateSuccessfulErrorFlowDoesNotThrowErrors() { + TestApplicationContext testApplicationContext = TestUtils.createTestApplicationContext(); + testApplicationContext.refresh(); DirectChannel outChannel = new DirectChannel(); outChannel.subscribe(message -> { throw new RuntimeException("problems"); @@ -90,13 +91,13 @@ public class MessageProducerSupportTests { PublishSubscribeChannel errorChannel = new PublishSubscribeChannel(); SuccessfulErrorService errorService = new SuccessfulErrorService(); ServiceActivatingHandler handler = new ServiceActivatingHandler(errorService); - handler.setBeanFactory(mock(BeanFactory.class)); + handler.setBeanFactory(testApplicationContext); handler.afterPropertiesSet(); errorChannel.subscribe(handler); MessageProducerSupport mps = new MessageProducerSupport() { }; mps.setOutputChannel(outChannel); mps.setErrorChannel(errorChannel); - mps.setBeanFactory(TestUtils.createTestApplicationContext()); + mps.setBeanFactory(testApplicationContext); mps.afterPropertiesSet(); mps.start(); Message message = new GenericMessage("hello"); @@ -106,6 +107,7 @@ public class MessageProducerSupportTests { assertEquals(MessageDeliveryException.class, errorMessage.getPayload().getClass()); MessageDeliveryException exception = (MessageDeliveryException) errorMessage.getPayload(); assertEquals(message, exception.getFailedMessage()); + testApplicationContext.close(); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ServiceActivatorEndpointTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ServiceActivatorEndpointTests.java index ca330912b8..5530bda950 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ServiceActivatorEndpointTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ServiceActivatorEndpointTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -84,13 +84,15 @@ public class ServiceActivatorEndpointTests { @Test public void returnAddressHeaderWithChannelName() { + TestUtils.TestApplicationContext testApplicationContext = TestUtils.createTestApplicationContext(); + testApplicationContext.refresh(); QueueChannel channel = new QueueChannel(1); channel.setBeanName("testChannel"); TestChannelResolver channelResolver = new TestChannelResolver(); channelResolver.addChannel("testChannel", channel); ServiceActivatingHandler endpoint = this.createEndpoint(); endpoint.setChannelResolver(channelResolver); - endpoint.setBeanFactory(mock(BeanFactory.class)); + endpoint.setBeanFactory(testApplicationContext); endpoint.afterPropertiesSet(); Message message = MessageBuilder.withPayload("foo") .setReplyChannelName("testChannel").build(); @@ -98,10 +100,13 @@ public class ServiceActivatorEndpointTests { Message reply = channel.receive(0); assertNotNull(reply); assertEquals("FOO", reply.getPayload()); + testApplicationContext.close(); } @Test public void dynamicReplyChannel() throws Exception { + TestUtils.TestApplicationContext testApplicationContext = TestUtils.createTestApplicationContext(); + testApplicationContext.refresh(); final QueueChannel replyChannel1 = new QueueChannel(); final QueueChannel replyChannel2 = new QueueChannel(); replyChannel2.setBeanName("replyChannel2"); @@ -116,7 +121,7 @@ public class ServiceActivatorEndpointTests { TestChannelResolver channelResolver = new TestChannelResolver(); channelResolver.addChannel("replyChannel2", replyChannel2); endpoint.setChannelResolver(channelResolver); - endpoint.setBeanFactory(mock(BeanFactory.class)); + endpoint.setBeanFactory(testApplicationContext); endpoint.afterPropertiesSet(); Message testMessage1 = MessageBuilder.withPayload("bar") .setReplyChannel(replyChannel1).build(); @@ -134,6 +139,7 @@ public class ServiceActivatorEndpointTests { reply2 = replyChannel2.receive(0); assertNotNull(reply2); assertEquals("foobar", reply2.getPayload()); + testApplicationContext.close(); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/filter/FilterAnnotationMethodResolutionTests.java b/spring-integration-core/src/test/java/org/springframework/integration/filter/FilterAnnotationMethodResolutionTests.java index 8c31138239..c4d2485b35 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/filter/FilterAnnotationMethodResolutionTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/filter/FilterAnnotationMethodResolutionTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -22,24 +22,28 @@ import static org.junit.Assert.assertTrue; import org.junit.Test; -import org.springframework.beans.factory.support.DefaultListableBeanFactory; -import org.springframework.messaging.Message; import org.springframework.integration.annotation.Filter; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.FilterFactoryBean; -import org.springframework.messaging.MessageHandler; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandler; /** * @author Mark Fisher + * @author Artme Bilan + * * @since 2.0 */ public class FilterAnnotationMethodResolutionTests { @Test public void resolveAnnotatedMethod() throws Exception { + TestUtils.TestApplicationContext testApplicationContext = TestUtils.createTestApplicationContext(); + testApplicationContext.refresh(); FilterFactoryBean factoryBean = new FilterFactoryBean(); - factoryBean.setBeanFactory(new DefaultListableBeanFactory()); + factoryBean.setBeanFactory(testApplicationContext); AnnotatedTestFilter filter = new AnnotatedTestFilter(); factoryBean.setTargetObject(filter); MessageHandler handler = factoryBean.getObject(); @@ -49,6 +53,7 @@ public class FilterAnnotationMethodResolutionTests { assertNotNull(result); assertTrue(filter.invokedCorrectMethod); assertFalse(filter.invokedIncorrectMethod); + testApplicationContext.close(); } @@ -78,6 +83,7 @@ public class FilterAnnotationMethodResolutionTests { this.invokedCorrectMethod = true; return true; } + } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/MessagingGatewayTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/MessagingGatewayTests.java index aa48d11618..319648dfc6 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/MessagingGatewayTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/MessagingGatewayTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -248,13 +248,15 @@ public class MessagingGatewayTests { // should not fail but it does now @Test public void validateErrorChannelWithSuccessfulReply() { + TestUtils.TestApplicationContext testApplicationContext = TestUtils.createTestApplicationContext(); + testApplicationContext.refresh(); DirectChannel reqChannel = new DirectChannel(); reqChannel.subscribe(message -> { throw new RuntimeException("ooops"); }); PublishSubscribeChannel errorChannel = new PublishSubscribeChannel(); ServiceActivatingHandler handler = new ServiceActivatingHandler(new MyOneWayErrorService()); - handler.setBeanFactory(mock(BeanFactory.class)); + handler.setBeanFactory(testApplicationContext); handler.afterPropertiesSet(); errorChannel.subscribe(handler); @@ -268,6 +270,7 @@ public class MessagingGatewayTests { this.messagingGateway.start(); this.messagingGateway.send("hello"); + testApplicationContext.close(); } public static class MyErrorService { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/MethodInvokingMessageProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/MethodInvokingMessageProcessorTests.java index bab5ec2ce6..af5c4c2508 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/MethodInvokingMessageProcessorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/MethodInvokingMessageProcessorTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -41,6 +41,7 @@ import org.junit.Test; import org.junit.rules.ExpectedException; import org.springframework.beans.factory.BeanFactory; +import org.springframework.expression.spel.SpelCompilerMode; import org.springframework.expression.spel.SpelEvaluationException; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.gateway.GatewayProxyFactoryBean; @@ -51,6 +52,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.handler.annotation.Header; import org.springframework.messaging.support.GenericMessage; +import org.springframework.util.StopWatch; /** @@ -317,6 +319,7 @@ public class MethodInvokingMessageProcessorTests { TestDifferentErrorService service = new TestDifferentErrorService(); Method method = TestErrorService.class.getMethod("checked", String.class); MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(service, method); + processor.setUseSpelInvoker(true); processor.processMessage(new GenericMessage("foo")); } @@ -567,7 +570,50 @@ public class MethodInvokingMessageProcessorTests { assertEquals("BAR", helper.process(new GenericMessage<>("bar"))); } + @Test + public void testPerformanceSpelVersusInvocable() throws Exception { + AnnotatedTestService service = new AnnotatedTestService(); + Method method = service.getClass().getMethod("messageAndHeader", Message.class, Integer.class); + + MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(service, method); + processor.setUseSpelInvoker(true); + + Message message = MessageBuilder.withPayload("foo").setHeader("number", 42).build(); + + StopWatch stopWatch = new StopWatch("SpEL vs Invocable Performance"); + + stopWatch.start("SpEL"); + for (int i = 0; i < 10000; i++) { + processor.processMessage(message); + } + stopWatch.stop(); + + processor = new MethodInvokingMessageProcessor(service, method); + + stopWatch.start("Invocable"); + for (int i = 0; i < 10000; i++) { + processor.processMessage(message); + } + stopWatch.stop(); + + + System.setProperty("spring.expression.compiler.mode", SpelCompilerMode.IMMEDIATE.name()); + processor = new MethodInvokingMessageProcessor(service, method); + processor.setUseSpelInvoker(true); + + stopWatch.start("Compiled SpEL"); + for (int i = 0; i < 10000; i++) { + processor.processMessage(message); + } + stopWatch.stop(); + + System.clearProperty("spring.expression.compiler.mode"); + + logger.warn(stopWatch.prettyPrint()); + } + private static class ExceptionCauseMatcher extends TypeSafeMatcher { + private Throwable cause; private final Class type; @@ -578,7 +624,6 @@ public class MethodInvokingMessageProcessorTests { @Override public boolean matchesSafely(Exception item) { - logger.debug(item); cause = item.getCause(); assertNotNull("There is no cause for " + item, cause); return type.isAssignableFrom(cause.getClass()); @@ -588,6 +633,7 @@ public class MethodInvokingMessageProcessorTests { public void describeTo(Description description) { description.appendText("cause to be ").appendValue(type).appendText("but was ").appendValue(cause); } + } @SuppressWarnings("unused") @@ -617,6 +663,7 @@ public class MethodInvokingMessageProcessorTests { public String checked(String input) throws Exception { throw new CheckedException("Expected test exception"); } + } @SuppressWarnings("serial") diff --git a/spring-integration-core/src/test/java/org/springframework/integration/message/PayloadAndHeaderMappingTests.java b/spring-integration-core/src/test/java/org/springframework/integration/message/PayloadAndHeaderMappingTests.java index dc6e80ae57..5a4be93fa7 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/message/PayloadAndHeaderMappingTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/message/PayloadAndHeaderMappingTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -27,11 +27,15 @@ import java.util.HashMap; import java.util.Map; import java.util.Properties; +import org.junit.AfterClass; import org.junit.Before; +import org.junit.BeforeClass; import org.junit.Test; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.integration.handler.ServiceActivatingHandler; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.handler.annotation.Header; @@ -40,13 +44,26 @@ import org.springframework.messaging.handler.annotation.Payload; /** * @author Mark Fisher + * @author Artem Bilan + * * @since 1.0.3 */ @SuppressWarnings({ "rawtypes", "unchecked" }) public class PayloadAndHeaderMappingTests { + private static final ConfigurableApplicationContext applicationContext = TestUtils.createTestApplicationContext(); + private TestBean bean; + @BeforeClass + public static void start() { + applicationContext.refresh(); + } + + @AfterClass + public static void stop() { + applicationContext.close(); + } @Before public void setup() { @@ -678,8 +695,8 @@ public class PayloadAndHeaderMappingTests { public void twoPayloadExpressions() throws Exception { MessageHandler handler = this.getHandler("twoPayloadExpressions", String.class, String.class); Map payload = new HashMap(); - payload.put("foo", new Integer(123)); - payload.put("bar", new Integer(456)); + payload.put("foo", 123); + payload.put("bar", 456); Message message = MessageBuilder.withPayload(payload).build(); handler.handleMessage(message); assertNull(bean.lastHeaders); @@ -689,7 +706,11 @@ public class PayloadAndHeaderMappingTests { private ServiceActivatingHandler getHandler(String methodName, Class... types) throws Exception { - return new ServiceActivatingHandler(bean, TestBean.class.getMethod(methodName, types)); + ServiceActivatingHandler serviceActivatingHandler = + new ServiceActivatingHandler(bean, TestBean.class.getMethod(methodName, types)); + serviceActivatingHandler.setBeanFactory(applicationContext); + serviceActivatingHandler.afterPropertiesSet(); + return serviceActivatingHandler; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/config/RouterFactoryBeanTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/config/RouterFactoryBeanTests.java index cb487b56c2..d277666da9 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/router/config/RouterFactoryBeanTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/config/RouterFactoryBeanTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,20 +18,18 @@ package org.springframework.integration.router.config; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; -import static org.mockito.Mockito.doReturn; -import static org.mockito.Mockito.mock; import org.junit.Test; -import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.RouterFactoryBean; -import org.springframework.messaging.MessageChannel; +import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.support.GenericMessage; /** * @author Gary Russell + * @author Artem Bilan * @since 4.2.5 * */ @@ -41,19 +39,21 @@ public class RouterFactoryBeanTests { @Test public void testOutputChannelName() throws Exception { + TestUtils.TestApplicationContext testApplicationContext = TestUtils.createTestApplicationContext(); + testApplicationContext.refresh(); RouterFactoryBean fb = new RouterFactoryBean(); fb.setTargetObject(this); fb.setTargetMethodName("foo"); fb.setDefaultOutputChannelName("bar"); - BeanFactory beanFactory = mock(BeanFactory.class); QueueChannel bar = new QueueChannel(); - doReturn(bar).when(beanFactory).getBean("bar", MessageChannel.class); - fb.setBeanFactory(beanFactory); + testApplicationContext.registerBean("bar", bar); + fb.setBeanFactory(testApplicationContext); MessageHandler handler = fb.getObject(); this.routeAttempted = false; handler.handleMessage(new GenericMessage<>("foo")); assertNotNull(bar.receive(10000)); assertTrue(this.routeAttempted); + testApplicationContext.close(); } public String foo() { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/util/BeanFactoryTypeConverterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/util/BeanFactoryTypeConverterTests.java index c9d8243e18..dca2a15db3 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/util/BeanFactoryTypeConverterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/util/BeanFactoryTypeConverterTests.java @@ -68,6 +68,7 @@ import org.springframework.messaging.support.GenericMessage; * @author Oleg Zhurakousky * @author Gary Russell * @author Gunnar Hillert + * @author Artem Bilan * */ public class BeanFactoryTypeConverterTests { @@ -78,7 +79,9 @@ public class BeanFactoryTypeConverterTests { BeanFactoryTypeConverter typeConverter = new BeanFactoryTypeConverter(); List sourceObject = new ArrayList(); ArrayList convertedCollection = - (ArrayList) typeConverter.convertValue(sourceObject, TypeDescriptor.forObject(sourceObject), TypeDescriptor.forObject(new ArrayList())); + (ArrayList) typeConverter.convertValue(sourceObject, + TypeDescriptor.forObject(sourceObject), + TypeDescriptor.forObject(new ArrayList())); assertEquals(sourceObject, convertedCollection); } @@ -86,7 +89,8 @@ public class BeanFactoryTypeConverterTests { public void testToStringConversion() { BeanFactoryTypeConverter typeConverter = new BeanFactoryTypeConverter(); typeConverter.setBeanFactory(new DefaultListableBeanFactory()); - String converted = (String) typeConverter.convertValue(new Integer(1234), TypeDescriptor.valueOf(Integer.class), TypeDescriptor.valueOf(String.class)); + String converted = (String) typeConverter.convertValue(1234, TypeDescriptor.valueOf(Integer.class), + TypeDescriptor.valueOf(String.class)); assertEquals("1234", converted); } @@ -95,7 +99,9 @@ public class BeanFactoryTypeConverterTests { BeanFactoryTypeConverter typeConverter = new BeanFactoryTypeConverter(); typeConverter.setBeanFactory(new DefaultListableBeanFactory()); @SuppressWarnings("unchecked") - Collection converted = (Collection) typeConverter.convertValue(new Integer(1234), TypeDescriptor.valueOf(Integer.class), TypeDescriptor.forObject(new ArrayList(Arrays.asList(1)))); + Collection converted = (Collection) typeConverter.convertValue(1234, + TypeDescriptor.valueOf(Integer.class), + TypeDescriptor.forObject(new ArrayList(Arrays.asList(1)))); assertEquals(Arrays.asList(1234), converted); } @@ -104,7 +110,8 @@ public class BeanFactoryTypeConverterTests { BeanFactoryTypeConverter typeConverter = new BeanFactoryTypeConverter(); typeConverter.setBeanFactory(new DefaultListableBeanFactory()); MessageHeaders headers = new GenericMessage("foo").getHeaders(); - assertSame(headers, typeConverter.convertValue(headers, TypeDescriptor.valueOf(MessageHeaders.class), TypeDescriptor.valueOf(MessageHeaders.class))); + assertSame(headers, typeConverter.convertValue(headers, TypeDescriptor.valueOf(MessageHeaders.class), + TypeDescriptor.valueOf(MessageHeaders.class))); } @Test @@ -124,7 +131,8 @@ public class BeanFactoryTypeConverterTests { } }); MessageHistory history = MessageHistory.read(message); - assertSame(history, typeConverter.convertValue(history, TypeDescriptor.valueOf(MessageHistory.class), TypeDescriptor.valueOf(MessageHistory.class))); + assertSame(history, typeConverter.convertValue(history, TypeDescriptor.valueOf(MessageHistory.class), + TypeDescriptor.valueOf(MessageHistory.class))); } @Test @@ -132,7 +140,8 @@ public class BeanFactoryTypeConverterTests { BeanFactoryTypeConverter typeConverter = new BeanFactoryTypeConverter(); typeConverter.setBeanFactory(new DefaultListableBeanFactory()); byte[] bytes = new byte[1]; - assertSame(bytes, typeConverter.convertValue(bytes, TypeDescriptor.valueOf(byte[].class), TypeDescriptor.valueOf(byte[].class))); + assertSame(bytes, typeConverter.convertValue(bytes, TypeDescriptor.valueOf(byte[].class), + TypeDescriptor.valueOf(byte[].class))); } @Test @@ -140,7 +149,8 @@ public class BeanFactoryTypeConverterTests { BeanFactoryTypeConverter typeConverter = new BeanFactoryTypeConverter(); typeConverter.setBeanFactory(new DefaultListableBeanFactory()); String string = "foo"; - assertSame(string, typeConverter.convertValue(string, TypeDescriptor.valueOf(String.class), TypeDescriptor.valueOf(Object.class))); + assertSame(string, typeConverter.convertValue(string, TypeDescriptor.valueOf(String.class), + TypeDescriptor.valueOf(Object.class))); } @Test @@ -153,7 +163,8 @@ public class BeanFactoryTypeConverterTests { BeanFactoryTypeConverter typeConverter = new BeanFactoryTypeConverter(conversionService); typeConverter.setBeanFactory(new DefaultListableBeanFactory()); Object object = new Object(); - assertEquals("foo", typeConverter.convertValue(object, TypeDescriptor.valueOf(Object.class), TypeDescriptor.valueOf(String.class))); + assertEquals("foo", typeConverter.convertValue(object, TypeDescriptor.valueOf(Object.class), + TypeDescriptor.valueOf(String.class))); } @SuppressWarnings("unchecked") @@ -189,8 +200,9 @@ public class BeanFactoryTypeConverterTests { assertThat(bars.get("foo").get("foo").iterator().next(), instanceOf(Bar.class)); Service service = new Service(); - MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(service, "handle"); + MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor<>(service, "handle"); processor.setConversionService(conversionService); + processor.setUseSpelInvoker(true); ServiceActivatingHandler handler = new ServiceActivatingHandler(processor); QueueChannel replyChannel = new QueueChannel(); handler.setOutputChannel(replyChannel); @@ -215,13 +227,14 @@ public class BeanFactoryTypeConverterTests { typeConverter.setBeanFactory(beanFactory); Service service = new Service(); - MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(service, "handle"); + MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor<>(service, "handle"); processor.setConversionService(conversionService); + processor.setUseSpelInvoker(true); ServiceActivatingHandler handler = new ServiceActivatingHandler(processor); QueueChannel replyChannel = new QueueChannel(); handler.setOutputChannel(replyChannel); handler.handleMessage(new GenericMessage>(Collections.singletonList(new Foo()))); - Message message = replyChannel.receive(0); + Message message = replyChannel.receive(10000); assertNotNull(message); assertEquals("baz", message.getPayload()); } @@ -334,5 +347,7 @@ public class BeanFactoryTypeConverterTests { assertThat(payload.iterator().next(), instanceOf(Bar.class)); return "baz"; } + } + } diff --git a/spring-integration-test/src/main/java/org/springframework/integration/test/util/TestUtils.java b/spring-integration-test/src/main/java/org/springframework/integration/test/util/TestUtils.java index 7d3aa1e7f5..0d4ccd7aae 100644 --- a/spring-integration-test/src/main/java/org/springframework/integration/test/util/TestUtils.java +++ b/spring-integration-test/src/main/java/org/springframework/integration/test/util/TestUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -35,6 +35,7 @@ import org.springframework.beans.factory.BeanNameAware; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.context.support.GenericApplicationContext; +import org.springframework.format.support.DefaultFormattingConversionService; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessagingException; @@ -90,6 +91,7 @@ public abstract class TestUtils { ThreadPoolTaskScheduler scheduler = createTaskScheduler(10); scheduler.setErrorHandler(errorHandler); registerBean("taskScheduler", scheduler, context); + registerBean("integrationConversionService", new DefaultFormattingConversionService(), context); return context; }