From 09fb4f78c9ffce753abd035f7754764d0a61ffa4 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Sun, 5 Jul 2015 12:58:33 -0400 Subject: [PATCH] INT-2166: Add SecurityContext Propagation JIRA: https://jira.spring.io/browse/INT-2166 * Introduce `ThreadStatePropagationChannelInterceptor` based on the `ExecutorChannelInterceptor` * Add `SecurityContextPropagationChannelInterceptor`,`SecurityContextCleanupChannelInterceptor` * Introduce `AbstractExecutorChannel` to utilize `ExecutorChannelInterceptor` logic * Introduce `MessageHandlingTaskDecorator` to avoid package tangle from `dispatcher` and `channel` * Introduce `SecurityContextCleanupAdvice` for those cases when we don't get deal with `MessageChannel`s already, but want to have proper way to cleanup `SecurityContext` * Make `GlobalChannelInterceptorProcessor` as `SmartInitializingSingleton` to avoid `phase` conflicts. * Fix `MessagingAnnotationPostProcessor` to use `beanFactory.initializeBean(endpoint, endpointBeanName);` instead of manual `start()` invocation bypassing the `phase` logic, hence having a bug, when endpoints have been started very early * Optimise `AbstractPollableChannel` to use `size` field from `ChannelInterceptorList` instead of `size()` from `Collection` * Fix `AnnotatedEndpointActivationTests` extracting separate component for annotation configuration instead of using test class directly. This caused very late Messaging Annotations process on that class * Fix typo in the `spring-integration-jdbc-4.2.xsd` * Remove some `SOUT`s throughout the project TODO Docs PR Comments: * Remove redundant `AbstractExecutorChannel#executorInterceptors` and make logic based on the `super.interceptors` * Fix wrong imports order * JavaDocs for `ThreadStatePropagationChannelInterceptor` * Docs for `SecurityContext` propagation INT-3593: Fix FTP PartialSuccess Tests JIRA: https://jira.spring.io/browse/INT-3593 Sort the files for the MPUT tests. INT-2166: Add SecurityContext Propagation JIRA: https://jira.spring.io/browse/INT-2166 * Introduce `ThreadStatePropagationChannelInterceptor` based on the `ExecutorChannelInterceptor` * Add `SecurityContextPropagationChannelInterceptor`,`SecurityContextCleanupChannelInterceptor` * Introduce `AbstractExecutorChannel` to utilize `ExecutorChannelInterceptor` logic * Introduce `MessageHandlingTaskDecorator` to avoid package tangle from `dispatcher` and `channel` * Introduce `SecurityContextCleanupAdvice` for those cases when we don't get deal with `MessageChannel`s already, but want to have proper way to cleanup `SecurityContext` * Make `GlobalChannelInterceptorProcessor` as `SmartInitializingSingleton` to avoid `phase` conflicts. * Fix `MessagingAnnotationPostProcessor` to use `beanFactory.initializeBean(endpoint, endpointBeanName);` instead of manual `start()` invocation bypassing the `phase` logic, hence having a bug, when endpoints have been started very early * Optimise `AbstractPollableChannel` to use `size` field from `ChannelInterceptorList` instead of `size()` from `Collection` * Fix `AnnotatedEndpointActivationTests` extracting separate component for annotation configuration instead of using test class directly. This caused very late Messaging Annotations process on that class * Fix typo in the `spring-integration-jdbc-4.2.xsd` * Remove some `SOUT`s throughout the project TODO Docs PR Comments: * Remove redundant `AbstractExecutorChannel#executorInterceptors` and make logic based on the `super.interceptors` * Fix wrong imports order * JavaDocs for `ThreadStatePropagationChannelInterceptor` * Docs for `SecurityContext` propagation Doc Polishing Address PR comments Address PR comments * Extract `ExecutorChannelInterceptor` logic in the `PollingConsumer` to have an ability to invoke `afterMessageHandled()` on the TaskScheduler's Thread for example for the `SecurityContext` clean up * Get rid of all that redundant "clean up" stuff * Docs polishing Fix `NPE` in the `PollingConsumer` Introduce `ExecutorChannelInterceptorAware` to avoid iterators on each message Polishing; Docs, Sonar --- build.gradle | 5 +- .../amqp/channel/PollableAmqpChannel.java | 57 ++++- .../channel/AbstractExecutorChannel.java | 209 ++++++++++++++++++ .../channel/AbstractMessageChannel.java | 4 +- .../channel/AbstractPollableChannel.java | 64 +++++- .../integration/channel/ExecutorChannel.java | 66 +++--- .../ExecutorChannelInterceptorAware.java | 33 +++ .../channel/PublishSubscribeChannel.java | 63 +++--- ...eadStatePropagationChannelInterceptor.java | 129 +++++++++++ .../GlobalChannelInterceptorProcessor.java | 76 +++---- ...AbstractMethodAnnotationPostProcessor.java | 1 + .../MessagingAnnotationPostProcessor.java | 91 ++------ .../dispatcher/BroadcastingDispatcher.java | 70 +++++- .../MessageHandlingTaskDecorator.java | 34 +++ .../dispatcher/UnicastingDispatcher.java | 64 +++++- .../integration/endpoint/PollingConsumer.java | 102 ++++++++- .../channel/ExecutorChannelTests.java | 100 ++++++++- .../interceptor/ChannelInterceptorTests.java | 70 +++++- ...notatedEndpointActivationTests-context.xml | 2 + .../AnnotatedEndpointActivationTests.java | 40 ++-- .../config/spring-integration-jdbc-4.2.xsd | 2 +- .../integration/jms/PollableJmsChannel.java | 58 ++++- ...eplyScenariosWithCachedConsumersTests.java | 3 +- .../jmx/MBeanAttributeFilterTests.java | 1 - ...onPublishingChannelAdapterParserTests.java | 2 - .../security/channel/ChannelAccessPolicy.java | 2 +- .../channel/ChannelSecurityInterceptor.java | 1 + ...yContextPropagationChannelInterceptor.java | 95 ++++++++ ...dapterSecurityIntegrationTests-context.xml | 35 ++- ...hannelAdapterSecurityIntegrationTests.java | 64 ++++-- .../ChannelSecurityInterceptorTests.java | 2 +- ...erceptorSecuredChannelAnnotationTests.java | 116 ++++++++++ .../config/SecuredChannelsParserTests.java | 2 +- .../src/test/resources/log4j.properties | 2 +- .../WebServiceInboundGatewayParserTests.java | 3 +- src/reference/asciidoc/security.adoc | 70 +++++- src/reference/asciidoc/whats-new.adoc | 9 +- 37 files changed, 1459 insertions(+), 288 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannelInterceptorAware.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ThreadStatePropagationChannelInterceptor.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageHandlingTaskDecorator.java create mode 100644 spring-integration-security/src/main/java/org/springframework/integration/security/channel/SecurityContextPropagationChannelInterceptor.java diff --git a/build.gradle b/build.gradle index 86a36c692f..92bb67184a 100644 --- a/build.gradle +++ b/build.gradle @@ -551,9 +551,8 @@ project('spring-integration-security') { compile("org.springframework.security:spring-security-core:$springSecurityVersion") { exclude group: 'org.springframework', module: 'spring-support' } - compile("org.springframework.security:spring-security-config:$springSecurityVersion") { - exclude group: 'org.springframework', module: 'spring-support' - } + + testCompile "org.springframework.security:spring-security-config:$springSecurityVersion" } } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java index 1b134615a5..f064be3671 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java @@ -18,16 +18,19 @@ package org.springframework.integration.amqp.channel; import java.util.ArrayDeque; import java.util.Deque; +import java.util.List; import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.integration.channel.ExecutorChannelInterceptorAware; import org.springframework.integration.channel.management.PollableChannelManagement; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.ChannelInterceptor; +import org.springframework.messaging.support.ExecutorChannelInterceptor; import org.springframework.util.Assert; /** @@ -40,8 +43,8 @@ import org.springframework.util.Assert; * @author Gary Russell * @since 2.1 */ -public class PollableAmqpChannel extends AbstractAmqpChannel implements PollableChannel, - PollableChannelManagement { +public class PollableAmqpChannel extends AbstractAmqpChannel + implements PollableChannel, PollableChannelManagement, ExecutorChannelInterceptorAware { private final String channelName; @@ -49,6 +52,7 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable private volatile AmqpAdmin amqpAdmin; + private volatile int executorInterceptorsSize; public PollableAmqpChannel(String channelName, AmqpTemplate amqpTemplate) { super(amqpTemplate); @@ -184,4 +188,53 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable return this.receive(); } + @Override + public void setInterceptors(List interceptors) { + super.setInterceptors(interceptors); + for (ChannelInterceptor interceptor : interceptors) { + if (interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize++; + } + } + } + + @Override + public void addInterceptor(ChannelInterceptor interceptor) { + super.addInterceptor(interceptor); + if (interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize++; + } + } + + @Override + public void addInterceptor(int index, ChannelInterceptor interceptor) { + super.addInterceptor(index, interceptor); + if (interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize++; + } + } + + @Override + public boolean removeInterceptor(ChannelInterceptor interceptor) { + boolean removed = super.removeInterceptor(interceptor); + if (removed && interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize--; + } + return removed; + } + + @Override + public ChannelInterceptor removeInterceptor(int index) { + ChannelInterceptor interceptor = super.removeInterceptor(index); + if (interceptor != null && interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize--; + } + return interceptor; + } + + @Override + public boolean hasExecutorInterceptors() { + return this.executorInterceptorsSize > 0; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java new file mode 100644 index 0000000000..da72a346f8 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java @@ -0,0 +1,209 @@ +/* + * Copyright 2015 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.channel; + +import java.util.ArrayDeque; +import java.util.Deque; +import java.util.Iterator; +import java.util.List; +import java.util.concurrent.Executor; + +import org.springframework.integration.dispatcher.AbstractDispatcher; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageDeliveryException; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.support.ChannelInterceptor; +import org.springframework.messaging.support.ExecutorChannelInterceptor; +import org.springframework.messaging.support.MessageHandlingRunnable; +import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; + +/** + * The {@link AbstractSubscribableChannel} base implementation for those inheritors + * which logic may be based on the {@link Executor}. + *

+ * Utilizes common operations for the {@link AbstractDispatcher}. + *

+ * Implements the {@link ExecutorChannelInterceptor}s logic when the message handling + * is handed to the {@link Executor#execute(Runnable)}. + * + * @author Artem Bilan + * @see ExecutorChannel + * @see PublishSubscribeChannel + * @since 4.2 + */ +public abstract class AbstractExecutorChannel extends AbstractSubscribableChannel + implements ExecutorChannelInterceptorAware { + + protected volatile Executor executor; + + protected volatile AbstractDispatcher dispatcher; + + protected volatile Integer maxSubscribers; + + protected volatile int executorInterceptorsSize; + + public AbstractExecutorChannel(Executor executor) { + this.executor = executor; + } + + /** + * Specify the maximum number of subscribers supported by the + * channel's dispatcher. + * + * @param maxSubscribers The maximum number of subscribers allowed. + */ + public void setMaxSubscribers(int maxSubscribers) { + this.maxSubscribers = maxSubscribers; + this.dispatcher.setMaxSubscribers(maxSubscribers); + } + + @Override + public void setInterceptors(List interceptors) { + super.setInterceptors(interceptors); + for (ChannelInterceptor interceptor : interceptors) { + if (interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize++; + } + } + } + + @Override + public void addInterceptor(ChannelInterceptor interceptor) { + super.addInterceptor(interceptor); + if (interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize++; + } + } + + @Override + public void addInterceptor(int index, ChannelInterceptor interceptor) { + super.addInterceptor(index, interceptor); + if (interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize++; + } + } + + @Override + public boolean removeInterceptor(ChannelInterceptor interceptor) { + boolean removed = super.removeInterceptor(interceptor); + if (removed && interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize--; + } + return removed; + } + + @Override + public ChannelInterceptor removeInterceptor(int index) { + ChannelInterceptor interceptor = super.removeInterceptor(index); + if (interceptor != null && interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize--; + } + return interceptor; + } + + @Override + public boolean hasExecutorInterceptors() { + return this.executorInterceptorsSize > 0; + } + + protected class MessageHandlingTask implements Runnable { + + private final MessageHandlingRunnable delegate; + + public MessageHandlingTask(MessageHandlingRunnable task) { + this.delegate = task; + } + + @Override + public void run() { + Message message = this.delegate.getMessage(); + MessageHandler messageHandler = this.delegate.getMessageHandler(); + Assert.notNull(messageHandler, "'messageHandler' must not be null"); + Deque interceptorStack = null; + try { + if (executorInterceptorsSize > 0) { + interceptorStack = new ArrayDeque(); + message = applyBeforeHandle(message, interceptorStack); + if (message == null) { + return; + } + } + messageHandler.handleMessage(message); + if (!CollectionUtils.isEmpty(interceptorStack)) { + triggerAfterMessageHandled(message, null, interceptorStack); + } + } + catch (Exception ex) { + if (!CollectionUtils.isEmpty(interceptorStack)) { + triggerAfterMessageHandled(message, ex, interceptorStack); + } + if (ex instanceof MessagingException) { + throw (MessagingException) ex; + } + String description = "Failed to handle " + message + " to " + this + " in " + messageHandler; + throw new MessageDeliveryException(message, description, ex); + } + catch (Error ex) {//NOSONAR - ok, we re-throw below + if (!CollectionUtils.isEmpty(interceptorStack)) { + String description = "Failed to handle " + message + " to " + this + " in " + messageHandler; + triggerAfterMessageHandled(message, new MessageDeliveryException(message, description, ex), + interceptorStack); + } + throw ex; + } + } + + private Message applyBeforeHandle(Message message, Deque interceptorStack) { + for (ChannelInterceptor interceptor : AbstractExecutorChannel.this.interceptors.interceptors) { + if (interceptor instanceof ExecutorChannelInterceptor) { + ExecutorChannelInterceptor executorInterceptor = (ExecutorChannelInterceptor) interceptor; + message = executorInterceptor.beforeHandle(message, AbstractExecutorChannel.this, + this.delegate.getMessageHandler()); + if (message == null) { + if (isLoggingEnabled() && logger.isDebugEnabled()) { + logger.debug(executorInterceptor.getClass().getSimpleName() + + " returned null from beforeHandle, i.e. precluding the send."); + } + triggerAfterMessageHandled(null, null, interceptorStack); + return null; + } + interceptorStack.add(executorInterceptor); + } + } + return message; + } + + private void triggerAfterMessageHandled(Message message, Exception ex, + Deque interceptorStack) { + Iterator iterator = interceptorStack.descendingIterator(); + while (iterator.hasNext()) { + ExecutorChannelInterceptor interceptor = iterator.next(); + try { + interceptor.afterMessageHandled(message, AbstractExecutorChannel.this, + this.delegate.getMessageHandler(), ex); + } + catch (Throwable ex2) {//NOSONAR + logger.error("Exception from afterMessageHandled in " + interceptor, ex2); + } + } + } + + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index 15f39fa51f..939ecd0d19 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -65,7 +65,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport implements MessageChannel, TrackableComponent, ChannelInterceptorAware, MessageChannelMetrics, ConfigurableMetricsAware { - private final ChannelInterceptorList interceptors; + protected final ChannelInterceptorList interceptors; private final Comparator orderComparator = new OrderComparator(); @@ -520,7 +520,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport private final Log logger; - private final List interceptors = new CopyOnWriteArrayList(); + protected final List interceptors = new CopyOnWriteArrayList(); private volatile int size; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java index 6798a64a33..9bdddbe368 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java @@ -18,11 +18,14 @@ package org.springframework.integration.channel; import java.util.ArrayDeque; import java.util.Deque; +import java.util.List; import org.springframework.integration.channel.management.PollableChannelManagement; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.ChannelInterceptor; +import org.springframework.messaging.support.ExecutorChannelInterceptor; +import org.springframework.util.CollectionUtils; /** * Base class for all pollable channels. @@ -30,10 +33,12 @@ import org.springframework.messaging.support.ChannelInterceptor; * @author Mark Fisher * @author Artem Bilan * @author Gary Russell + * @author Artem Bilan */ -public abstract class AbstractPollableChannel extends AbstractMessageChannel implements PollableChannel, - PollableChannelManagement { +public abstract class AbstractPollableChannel extends AbstractMessageChannel + implements PollableChannel, PollableChannelManagement, ExecutorChannelInterceptorAware { + protected volatile int executorInterceptorsSize; @Override public int getReceiveCount() { @@ -90,7 +95,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp if (logger.isTraceEnabled()) { logger.trace("preReceive on channel '" + this + "'"); } - if (interceptorList.getInterceptors().size() > 0) { + if (interceptorList.getSize() > 0) { interceptorStack = new ArrayDeque(); if (!interceptorList.preReceive(this, interceptorStack)) { @@ -108,7 +113,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp else if (logger.isTraceEnabled()) { logger.trace("postReceive on channel '" + this + "', message is null"); } - if (interceptorStack != null) { + if (!CollectionUtils.isEmpty(interceptorStack)) { message = interceptorList.postReceive(message, this); interceptorList.afterReceiveCompletion(message, this, null, interceptorStack); } @@ -118,13 +123,62 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp if (countsEnabled && !counted) { getMetrics().afterError(); } - if (interceptorStack != null) { + if (!CollectionUtils.isEmpty(interceptorStack)) { interceptorList.afterReceiveCompletion(null, this, e, interceptorStack); } throw e; } } + @Override + public void setInterceptors(List interceptors) { + super.setInterceptors(interceptors); + for (ChannelInterceptor interceptor : interceptors) { + if (interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize++; + } + } + } + + @Override + public void addInterceptor(ChannelInterceptor interceptor) { + super.addInterceptor(interceptor); + if (interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize++; + } + } + + @Override + public void addInterceptor(int index, ChannelInterceptor interceptor) { + super.addInterceptor(index, interceptor); + if (interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize++; + } + } + + @Override + public boolean removeInterceptor(ChannelInterceptor interceptor) { + boolean removed = super.removeInterceptor(interceptor); + if (removed && interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize--; + } + return removed; + } + + @Override + public ChannelInterceptor removeInterceptor(int index) { + ChannelInterceptor interceptor = super.removeInterceptor(index); + if (interceptor != null && interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize--; + } + return interceptor; + } + + @Override + public boolean hasExecutorInterceptors() { + return this.executorInterceptorsSize > 0; + } + /** * Subclasses must implement this method. A non-negative timeout indicates * how long to wait if the channel is empty (if the value is 0, it must diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java index 6eb9ff0305..484f7c84aa 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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,11 +20,13 @@ import java.util.concurrent.Executor; import org.springframework.integration.context.IntegrationProperties; import org.springframework.integration.dispatcher.LoadBalancingStrategy; +import org.springframework.integration.dispatcher.MessageHandlingTaskDecorator; import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; import org.springframework.integration.dispatcher.UnicastingDispatcher; import org.springframework.integration.support.channel.BeanFactoryChannelResolver; import org.springframework.integration.util.ErrorHandlingTaskExecutor; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.MessageHandlingRunnable; import org.springframework.util.Assert; import org.springframework.util.ErrorHandler; @@ -45,25 +47,17 @@ import org.springframework.util.ErrorHandler; * @author Artem Bilan * @since 1.0.3 */ -public class ExecutorChannel extends AbstractSubscribableChannel { - - private volatile UnicastingDispatcher dispatcher; - - private volatile Executor executor; +public class ExecutorChannel extends AbstractExecutorChannel { private volatile boolean failover = true; - private volatile Integer maxSubscribers; - private volatile LoadBalancingStrategy loadBalancingStrategy; - /** * Create an ExecutorChannel that delegates to the provided * {@link Executor} when dispatching Messages. *

* The Executor must not be null. - * * @param executor The executor. */ public ExecutorChannel(Executor executor) { @@ -75,46 +69,34 @@ public class ExecutorChannel extends AbstractSubscribableChannel { * delegates to the provided {@link Executor} when dispatching Messages. *

* The Executor must not be null. - * * @param executor The executor. * @param loadBalancingStrategy The load balancing strategy implementation. */ public ExecutorChannel(Executor executor, LoadBalancingStrategy loadBalancingStrategy) { + super(executor); Assert.notNull(executor, "executor must not be null"); - this.executor = executor; - this.dispatcher = new UnicastingDispatcher(executor); + UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher(executor); if (loadBalancingStrategy != null) { this.loadBalancingStrategy = loadBalancingStrategy; - this.dispatcher.setLoadBalancingStrategy(loadBalancingStrategy); + unicastingDispatcher.setLoadBalancingStrategy(loadBalancingStrategy); } + this.dispatcher = unicastingDispatcher; } /** * Specify whether the channel's dispatcher should have failover enabled. * By default, it will. Set this value to 'false' to disable it. - * * @param failover The failover boolean. */ public void setFailover(boolean failover) { this.failover = failover; - this.dispatcher.setFailover(failover); - } - - /** - * Specify the maximum number of subscribers supported by the - * channel's dispatcher. - * - * @param maxSubscribers The maximum number of subscribers allowed. - */ - public void setMaxSubscribers(int maxSubscribers) { - this.maxSubscribers = maxSubscribers; - this.dispatcher.setMaxSubscribers(maxSubscribers); + getDispatcher().setFailover(failover); } @Override protected UnicastingDispatcher getDispatcher() { - return this.dispatcher; + return (UnicastingDispatcher) this.dispatcher; } @Override @@ -124,15 +106,33 @@ public class ExecutorChannel extends AbstractSubscribableChannel { new BeanFactoryChannelResolver(this.getBeanFactory())); this.executor = new ErrorHandlingTaskExecutor(this.executor, errorHandler); } - this.dispatcher = new UnicastingDispatcher(this.executor); - this.dispatcher.setFailover(this.failover); + UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher(this.executor); + unicastingDispatcher.setFailover(this.failover); if (this.maxSubscribers == null) { - this.maxSubscribers = this.getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS, Integer.class); + this.maxSubscribers = + getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS, Integer.class); } - this.dispatcher.setMaxSubscribers(this.maxSubscribers); + unicastingDispatcher.setMaxSubscribers(this.maxSubscribers); if (this.loadBalancingStrategy != null) { - this.dispatcher.setLoadBalancingStrategy(this.loadBalancingStrategy); + unicastingDispatcher.setLoadBalancingStrategy(this.loadBalancingStrategy); } + + unicastingDispatcher.setMessageHandlingTaskDecorator(new MessageHandlingTaskDecorator() { + + @Override + public Runnable decorate(MessageHandlingRunnable task) { + if (ExecutorChannel.this.executorInterceptorsSize > 0) { + return new MessageHandlingTask(task); + } + else { + return task; + } + } + + }); + + this.dispatcher = unicastingDispatcher; } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannelInterceptorAware.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannelInterceptorAware.java new file mode 100644 index 0000000000..3ac4ff46fa --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannelInterceptorAware.java @@ -0,0 +1,33 @@ +/* + * Copyright 2015 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.channel; + +/** + * The {@link ChannelInterceptorAware} extension for the cases when + * the {@link org.springframework.messaging.support.ExecutorChannelInterceptor}s + * may have reason (e.g. {@link ExecutorChannel} or {@link QueueChannel}) + * and the implementors require to know if they should make the + * {@link org.springframework.messaging.support.ExecutorChannelInterceptor} + * or not. + * @author Artem Bilan + * @since 4.2 + */ +public interface ExecutorChannelInterceptorAware extends ChannelInterceptorAware { + + boolean hasExecutorInterceptors(); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java index 1a48fa4b37..ff76f65215 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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,8 +20,10 @@ import java.util.concurrent.Executor; import org.springframework.integration.context.IntegrationProperties; import org.springframework.integration.dispatcher.BroadcastingDispatcher; +import org.springframework.integration.dispatcher.MessageHandlingTaskDecorator; import org.springframework.integration.support.channel.BeanFactoryChannelResolver; import org.springframework.integration.util.ErrorHandlingTaskExecutor; +import org.springframework.messaging.support.MessageHandlingRunnable; import org.springframework.util.ErrorHandler; /** @@ -30,12 +32,9 @@ import org.springframework.util.ErrorHandler; * @author Mark Fisher * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan */ -public class PublishSubscribeChannel extends AbstractSubscribableChannel { - - private volatile BroadcastingDispatcher dispatcher; - - private volatile Executor executor; +public class PublishSubscribeChannel extends AbstractExecutorChannel { private volatile ErrorHandler errorHandler; @@ -45,9 +44,6 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel { private volatile int minSubscribers; - private volatile Integer maxSubscribers; - - /** * Create a PublishSubscribeChannel that will use an {@link Executor} * to invoke the handlers. If this is null, each invocation will occur in @@ -56,7 +52,7 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel { * @param executor The executor. */ public PublishSubscribeChannel(Executor executor) { - this.executor = executor; + super(executor); this.dispatcher = new BroadcastingDispatcher(executor); } @@ -98,12 +94,11 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel { * ignored. By default this is false meaning that an Exception * will be thrown whenever a handler fails. To override this and suppress * Exceptions, set the value to true. - * * @param ignoreFailures true if failures should be ignored. */ public void setIgnoreFailures(boolean ignoreFailures) { this.ignoreFailures = ignoreFailures; - this.getDispatcher().setIgnoreFailures(ignoreFailures); + getDispatcher().setIgnoreFailures(ignoreFailures); } /** @@ -113,23 +108,11 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel { * not be applied. If planning to use an Aggregator downstream * with the default correlation and completion strategies, you should set * this flag to true. - * * @param applySequence true if the sequence information should be applied. */ public void setApplySequence(boolean applySequence) { this.applySequence = applySequence; - this.getDispatcher().setApplySequence(applySequence); - } - - /** - * Specify the maximum number of subscribers supported by the - * channel's dispatcher. - * - * @param maxSubscribers The maximum number of subscribers allowed. - */ - public void setMaxSubscribers(int maxSubscribers) { - this.maxSubscribers = maxSubscribers; - this.getDispatcher().setMaxSubscribers(maxSubscribers); + getDispatcher().setApplySequence(applySequence); } /** @@ -141,7 +124,7 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel { */ public void setMinSubscribers(int minSubscribers) { this.minSubscribers = minSubscribers; - this.getDispatcher().setMinSubscribers(minSubscribers); + getDispatcher().setMinSubscribers(minSubscribers); } /** @@ -160,20 +143,36 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel { this.executor = new ErrorHandlingTaskExecutor(this.executor, this.errorHandler); } this.dispatcher = new BroadcastingDispatcher(this.executor); - this.dispatcher.setIgnoreFailures(this.ignoreFailures); - this.dispatcher.setApplySequence(this.applySequence); - this.dispatcher.setMinSubscribers(this.minSubscribers); + getDispatcher().setIgnoreFailures(this.ignoreFailures); + getDispatcher().setApplySequence(this.applySequence); + getDispatcher().setMinSubscribers(this.minSubscribers); } if (this.maxSubscribers == null) { - Integer maxSubscribers = this.getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS, Integer.class); + Integer maxSubscribers = + getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS, Integer.class); this.setMaxSubscribers(maxSubscribers); } - this.dispatcher.setBeanFactory(this.getBeanFactory()); + getDispatcher().setBeanFactory(this.getBeanFactory()); + + getDispatcher().setMessageHandlingTaskDecorator(new MessageHandlingTaskDecorator() { + + @Override + public Runnable decorate(MessageHandlingRunnable task) { + if (PublishSubscribeChannel.this.executorInterceptorsSize > 0) { + return new MessageHandlingTask(task); + } + else { + return task; + } + } + + }); + } @Override protected BroadcastingDispatcher getDispatcher() { - return this.dispatcher; + return (BroadcastingDispatcher) this.dispatcher; } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ThreadStatePropagationChannelInterceptor.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ThreadStatePropagationChannelInterceptor.java new file mode 100644 index 0000000000..04e41b4d2f --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ThreadStatePropagationChannelInterceptor.java @@ -0,0 +1,129 @@ +/* + * Copyright 2015 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.channel.interceptor; + +import java.io.Serializable; + +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.support.ChannelInterceptorAdapter; +import org.springframework.messaging.support.ExecutorChannelInterceptor; + +/** + * The {@link ExecutorChannelInterceptor} implementation responsible for + * the {@link Thread} (any?) state propagation from one message flow's thread to another + * through the {@link MessageChannel}s involved in the flow. + *

+ * The propagation is done from the {@link #preSend(Message, MessageChannel)} + * implementation using some internal {@link Message} extension which keeps the message + * to send and the state to propagate. + *

+ * The propagated state context extraction and population is done from the {@link #postReceive} + * implementation for the {@link org.springframework.messaging.PollableChannel}s, and from + * the {@link #beforeHandle} for the + * {@link org.springframework.integration.channel.AbstractExecutorChannel}s and + * {@link org.springframework.messaging.support.ExecutorSubscribableChannel}s + *

+ * Important. Any further interceptor, which modifies the message to send + * (e.g. {@code MessageBuilder.withPayload(...)...build()}), may drop the state to propagate. + * Such kind of interceptors combination should be revised properly. + * In most cases the interceptors reordering is enough to overcome the issue. + * + * @param the propagated state object type. + * + * @author Artem Bilan + * @since 4.2 + */ +public abstract class ThreadStatePropagationChannelInterceptor + extends ChannelInterceptorAdapter implements ExecutorChannelInterceptor { + + @Override + public final Message preSend(Message message, MessageChannel channel) { + S threadContext = obtainPropagatingContext(message, channel); + if (threadContext != null) { + return new MessageWithThreadState(message, threadContext); + } + else { + return message; + } + } + + @Override + @SuppressWarnings("unchecked") + public final Message postReceive(Message message, MessageChannel channel) { + if (message != null && message instanceof MessageWithThreadState) { + MessageWithThreadState messageWithThreadState = (MessageWithThreadState) message; + Message messageToHandle = messageWithThreadState.message; + populatePropagatedContext(messageWithThreadState.state, messageToHandle, channel); + return messageToHandle; + } + return message; + } + + @Override + public final Message beforeHandle(Message message, MessageChannel channel, MessageHandler handler) { + return postReceive(message, channel); + } + + @Override + public void afterMessageHandled(Message message, MessageChannel channel, MessageHandler handler, + Exception ex) { + // No-op + } + + protected abstract S obtainPropagatingContext(Message message, MessageChannel channel); + + protected abstract void populatePropagatedContext(S state, Message message, MessageChannel channel); + + + private static class MessageWithThreadState implements Message, Serializable { + + private static final long serialVersionUID = 1548216539234073073L; + + final Message message; + + final S state; + + public MessageWithThreadState(Message message, S state) { + this.message = message; + this.state = state; + } + + @Override + public Object getPayload() { + return this.message.getPayload(); + } + + @Override + public MessageHeaders getHeaders() { + return this.message.getHeaders(); + } + + @Override + public String toString() { + return "MessageWithThreadState{" + + "message=" + message + + ", state=" + state + + '}'; + } + + } + +} + diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/GlobalChannelInterceptorProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/GlobalChannelInterceptorProcessor.java index e39ed896be..277242ee9b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/GlobalChannelInterceptorProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/GlobalChannelInterceptorProcessor.java @@ -32,7 +32,7 @@ import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.ListableBeanFactory; -import org.springframework.context.SmartLifecycle; +import org.springframework.beans.factory.SmartInitializingSingleton; import org.springframework.core.OrderComparator; import org.springframework.integration.channel.ChannelInterceptorAware; import org.springframework.integration.channel.interceptor.GlobalChannelInterceptorWrapper; @@ -52,21 +52,21 @@ import org.springframework.util.StringUtils; * @author Gary Russell * @since 2.0 */ -final class GlobalChannelInterceptorProcessor implements BeanFactoryAware, SmartLifecycle { +final class GlobalChannelInterceptorProcessor implements BeanFactoryAware, SmartInitializingSingleton { private static final Log logger = LogFactory.getLog(GlobalChannelInterceptorProcessor.class); private final OrderComparator comparator = new OrderComparator(); - private final Set positiveOrderInterceptors = new LinkedHashSet(); + private final Set positiveOrderInterceptors = + new LinkedHashSet(); - private final Set negativeOrderInterceptors = new LinkedHashSet(); + private final Set negativeOrderInterceptors = + new LinkedHashSet(); private ListableBeanFactory beanFactory; - private volatile boolean processed; - @Override public void setBeanFactory(BeanFactory beanFactory) throws BeansException { Assert.isInstanceOf(ListableBeanFactory.class, beanFactory); @@ -74,51 +74,27 @@ final class GlobalChannelInterceptorProcessor implements BeanFactoryAware, Smart } @Override - public synchronized void start() { - if (!this.processed) { - Collection interceptors = this.beanFactory.getBeansOfType(GlobalChannelInterceptorWrapper.class).values(); - if (CollectionUtils.isEmpty(interceptors)) { - logger.debug("No global channel interceptors."); - } - else { - for (GlobalChannelInterceptorWrapper channelInterceptor : interceptors) { - if (channelInterceptor.getOrder() >= 0) { - this.positiveOrderInterceptors.add(channelInterceptor); - } - else { - this.negativeOrderInterceptors.add(channelInterceptor); - } - } - Map channels = this.beanFactory.getBeansOfType(ChannelInterceptorAware.class); - for (Entry entry : channels.entrySet()) { - this.addMatchingInterceptors(entry.getValue(), entry.getKey()); - } - } - this.processed = true; + public void afterSingletonsInstantiated() { + Collection interceptors = + this.beanFactory.getBeansOfType(GlobalChannelInterceptorWrapper.class).values(); + if (CollectionUtils.isEmpty(interceptors)) { + logger.debug("No global channel interceptors."); + } + else { + for (GlobalChannelInterceptorWrapper channelInterceptor : interceptors) { + if (channelInterceptor.getOrder() >= 0) { + this.positiveOrderInterceptors.add(channelInterceptor); + } + else { + this.negativeOrderInterceptors.add(channelInterceptor); + } + } + Map channels = + this.beanFactory.getBeansOfType(ChannelInterceptorAware.class); + for (Entry entry : channels.entrySet()) { + addMatchingInterceptors(entry.getValue(), entry.getKey()); + } } - } - - @Override - public void stop() { - } - - @Override - public boolean isRunning() { - return false; - } - - @Override - public int getPhase() { - return Integer.MIN_VALUE; - } - - @Override - public boolean isAutoStartup() { - return true; - } - - @Override - public void stop(Runnable callback) { } /** 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 60302199e6..55fbba8e7f 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 @@ -254,6 +254,7 @@ public abstract class AbstractMethodAnnotationPostProcessor, EnvironmentAware, - SmartInitializingSingleton { + InitializingBean, EnvironmentAware, SmartInitializingSingleton { private final Log logger = LogFactory.getLog(this.getClass()); @@ -91,12 +86,6 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean private final Map, MethodAnnotationPostProcessor> postProcessors = new HashMap, MethodAnnotationPostProcessor>(); - private final Set> listeners = new HashSet>(); - - private final Set lifecycles = new HashSet(); - - private volatile boolean running = true; - private final MultiValueMap lazyLifecyleRoles = new LinkedMultiValueMap(); @Override @@ -141,7 +130,7 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean } } catch (NoSuchBeanDefinitionException e) { - logger.error("No lifecyle role controller in context"); + logger.error("No LifecycleRoleController in the context"); } } @@ -154,8 +143,9 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean return bean; } ReflectionUtils.doWithMethods(beanClass, new ReflectionUtils.MethodCallback() { + @Override - @SuppressWarnings({ "unchecked", "rawtypes" }) + @SuppressWarnings({"unchecked", "rawtypes"}) public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { Map, List> annotationChains = new HashMap, List>(); @@ -181,7 +171,7 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean catch (NoSuchMethodException e) { throw new IllegalArgumentException("Service methods must be extracted to the service " + "interface for JdkDynamicProxy. The affected bean is: '" + beanName + "' " - + "and its method: '" + method + "'", e); + + "and its method: '" + method + "'", e); } } Object result = postProcessor.postProcess(bean, beanName, targetMethod, annotations); @@ -211,20 +201,8 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean String endpointBeanName = generateBeanName(beanName, method, annotationType); endpoint.setBeanName(endpointBeanName); beanFactory.registerSingleton(endpointBeanName, endpoint); - endpoint.setBeanFactory(beanFactory); - try { - endpoint.afterPropertiesSet(); - } - catch (Exception e) { - throw new BeanInitializationException("failed to initialize annotated component", e); - } - lifecycles.add(endpoint); - if (endpoint.isAutoStartup()) { - endpoint.start(); - } - if (result instanceof ApplicationListener) { - listeners.add((ApplicationListener) result); - } + beanFactory.initializeBean(endpoint, endpointBeanName); + Role role = AnnotationUtils.findAnnotation(method, Role.class); if (role != null) { lazyLifecyleRoles.add(role.value(), endpointBeanName); @@ -239,7 +217,7 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean } /** - * @param method the method. + * @param method the method. * @param annotationType the annotation type. * @return the hierarchical list of annotations in top-bottom order. */ @@ -258,7 +236,7 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean } private boolean recursiveFindAnnotation(Class annotationType, Annotation ann, - List annotationChain, Set visited) { + List annotationChain, Set visited) { if (ann.annotationType().equals(annotationType)) { annotationChain.add(ann); return true; @@ -281,8 +259,10 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean return (targetClass != null) ? targetClass : bean.getClass(); } - private String generateBeanName(String originalBeanName, Method method, Class annotationType) { - String baseName = originalBeanName + "." + method.getName() + "." + ClassUtils.getShortNameAsProperty(annotationType); + private String generateBeanName(String originalBeanName, Method method, + Class annotationType) { + String baseName = originalBeanName + "." + method.getName() + "." + + ClassUtils.getShortNameAsProperty(annotationType); String name = baseName; int count = 1; while (this.beanFactory.containsBean(name)) { @@ -291,47 +271,4 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean return name; } - - @Override - public void onApplicationEvent(ApplicationEvent event) { - for (ApplicationListener listener : listeners) { - try { - listener.onApplicationEvent(event); - } - catch (ClassCastException e) { - if (logger.isWarnEnabled() && event != null) { - logger.warn("ApplicationEvent of type [" + event.getClass() + - "] not accepted by ApplicationListener [" + listener + "]"); - } - } - } - } - - // Lifecycle implementation - - @Override - public boolean isRunning() { - return this.running; - } - - @Override - public void start() { - for (Lifecycle lifecycle : this.lifecycles) { - if (!lifecycle.isRunning()) { - lifecycle.start(); - } - } - this.running = true; - } - - @Override - public void stop() { - for (Lifecycle lifecycle : this.lifecycles) { - if (lifecycle.isRunning()) { - lifecycle.stop(); - } - } - this.running = false; - } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java index c984825140..179065e71a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java @@ -29,6 +29,8 @@ import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; +import org.springframework.messaging.support.MessageHandlingRunnable; +import org.springframework.util.Assert; /** * A broadcasting dispatcher implementation. If the 'ignoreFailures' property is set to false (the @@ -62,6 +64,16 @@ public class BroadcastingDispatcher extends AbstractDispatcher implements BeanFa private volatile boolean messageBuilderFactorySet; + private volatile MessageHandlingTaskDecorator messageHandlingTaskDecorator = + new MessageHandlingTaskDecorator() { + + @Override + public Runnable decorate(MessageHandlingRunnable task) { + return task; + } + + }; + private BeanFactory beanFactory; @@ -117,6 +129,11 @@ public class BroadcastingDispatcher extends AbstractDispatcher implements BeanFa this.minSubscribers = minSubscribers; } + public void setMessageHandlingTaskDecorator(MessageHandlingTaskDecorator messageHandlingTaskDecorator) { + Assert.notNull(messageHandlingTaskDecorator, "'messageHandlingTaskDecorator' must not be null."); + this.messageHandlingTaskDecorator = messageHandlingTaskDecorator; + } + @Override public void setBeanFactory(BeanFactory beanFactory) throws BeansException { this.beanFactory = beanFactory; @@ -141,16 +158,18 @@ public class BroadcastingDispatcher extends AbstractDispatcher implements BeanFa throw new MessageDispatchingException(message, "Dispatcher has no subscribers"); } int sequenceSize = handlers.size(); - for (final MessageHandler handler : handlers) { - final Message messageToSend = (!this.applySequence) ? message : getMessageBuilderFactory().fromMessage(message) - .pushSequenceDetails(message.getHeaders().getId(), sequenceNumber++, sequenceSize).build(); + for (MessageHandler handler : handlers) { + Message messageToSend = message; + if (this.applySequence) { + messageToSend = getMessageBuilderFactory() + .fromMessage(message) + .pushSequenceDetails(message.getHeaders().getId(), sequenceNumber++, sequenceSize) + .build(); + } + if (this.executor != null) { - this.executor.execute(new Runnable() { - @Override - public void run() { - invokeHandler(handler, messageToSend); - } - }); + Runnable task = createMessageHandlingTask(handler, messageToSend); + this.executor.execute(task); dispatched++; } else { @@ -170,6 +189,39 @@ public class BroadcastingDispatcher extends AbstractDispatcher implements BeanFa return dispatched >= minSubscribers; } + + private Runnable createMessageHandlingTask(final MessageHandler handler, final Message message) { + MessageHandlingRunnable task = new MessageHandlingRunnable() { + + final MessageHandler delegate = new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + invokeHandler(handler, message); + } + + }; + + @Override + public void run() { + invokeHandler(handler, message); + } + + @Override + public Message getMessage() { + return message; + } + + @Override + public MessageHandler getMessageHandler() { + return this.delegate; + } + + }; + + return this.messageHandlingTaskDecorator.decorate(task); + } + private boolean invokeHandler(MessageHandler handler, Message message) { try { handler.handleMessage(message); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageHandlingTaskDecorator.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageHandlingTaskDecorator.java new file mode 100644 index 0000000000..d55c9c786e --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageHandlingTaskDecorator.java @@ -0,0 +1,34 @@ +/* + * Copyright 2015 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.dispatcher; + +import org.springframework.messaging.support.MessageHandlingRunnable; + +/** + * The strategy to decorate {@link MessageHandlingRunnable} tasks + * to be used with the {@link java.util.concurrent.Executor}. + * + * @author Artem Bilan + * @since 4.2 + * @see UnicastingDispatcher + * @see BroadcastingDispatcher + */ +public interface MessageHandlingTaskDecorator { + + Runnable decorate(MessageHandlingRunnable task); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java index a2ac2cba2d..e394b6e690 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java @@ -1,4 +1,4 @@ -/* Copyright 2002-2014 the original author or authors. +/* Copyright 2002-2015 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. @@ -24,6 +24,9 @@ import org.springframework.integration.MessageDispatchingException; import org.springframework.messaging.Message; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.support.MessageHandlingRunnable; +import org.springframework.util.Assert; /** * Implementation of {@link MessageDispatcher} that will attempt to send a @@ -43,16 +46,35 @@ import org.springframework.messaging.MessageHandler; * @author Mark Fisher * @author Gary Russell * @author Oleg Zhurakousky + * @author Artem Bilan * @since 1.0.2 */ public class UnicastingDispatcher extends AbstractDispatcher { + private final MessageHandler dispatchHandler = new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + doDispatch(message); + } + + }; + + private final Executor executor; + private volatile boolean failover = true; private volatile LoadBalancingStrategy loadBalancingStrategy; - private final Executor executor; + private volatile MessageHandlingTaskDecorator messageHandlingTaskDecorator = + new MessageHandlingTaskDecorator() { + @Override + public Runnable decorate(MessageHandlingRunnable task) { + return task; + } + + }; public UnicastingDispatcher() { this.executor = null; @@ -83,20 +105,44 @@ public class UnicastingDispatcher extends AbstractDispatcher { this.loadBalancingStrategy = loadBalancingStrategy; } + public void setMessageHandlingTaskDecorator(MessageHandlingTaskDecorator messageHandlingTaskDecorator) { + Assert.notNull(messageHandlingTaskDecorator, "'messageHandlingTaskDecorator' must not be null."); + this.messageHandlingTaskDecorator = messageHandlingTaskDecorator; + } + @Override public final boolean dispatch(final Message message) { if (this.executor != null) { - this.executor.execute(new Runnable() { - @Override - public void run() { - doDispatch(message); - } - }); + Runnable task = createMessageHandlingTask(message); + this.executor.execute(task); return true; } return this.doDispatch(message); } + private Runnable createMessageHandlingTask(final Message message) { + MessageHandlingRunnable task = new MessageHandlingRunnable() { + + @Override + public void run() { + doDispatch(message); + } + + @Override + public Message getMessage() { + return message; + } + + @Override + public MessageHandler getMessageHandler() { + return UnicastingDispatcher.this.dispatchHandler; + } + + }; + + return this.messageHandlingTaskDecorator.decorate(task); + } + private boolean doDispatch(Message message) { if (this.tryOptimizedDispatch(message)) { return true; @@ -107,7 +153,7 @@ public class UnicastingDispatcher extends AbstractDispatcher { throw new MessageDispatchingException(message, "Dispatcher has no subscribers"); } List exceptions = new ArrayList(); - while (success == false && handlerIterator.hasNext()) { + while (!success && handlerIterator.hasNext()) { MessageHandler handler = handlerIterator.next(); try { handler.handleMessage(message); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java index 8cca2b4b68..1f02e5cf18 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2015 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. @@ -16,12 +16,23 @@ package org.springframework.integration.endpoint; +import java.util.ArrayDeque; +import java.util.Deque; +import java.util.Iterator; +import java.util.List; + import org.springframework.context.Lifecycle; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.PollableChannel; +import org.springframework.integration.channel.ExecutorChannelInterceptorAware; import org.springframework.integration.transaction.IntegrationResourceHolder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageDeliveryException; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.ChannelInterceptor; +import org.springframework.messaging.support.ExecutorChannelInterceptor; import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; /** * Message Endpoint that connects any {@link MessageHandler} implementation @@ -30,6 +41,7 @@ import org.springframework.util.Assert; * @author Mark Fisher * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan */ public class PollingConsumer extends AbstractPollingEndpoint { @@ -37,6 +49,8 @@ public class PollingConsumer extends AbstractPollingEndpoint { private final MessageHandler handler; + private final List channelInterceptors; + private volatile long receiveTimeout = 1000; public PollingConsumer(PollableChannel inputChannel, MessageHandler handler) { @@ -44,6 +58,12 @@ public class PollingConsumer extends AbstractPollingEndpoint { Assert.notNull(handler, "handler must not be null"); this.inputChannel = inputChannel; this.handler = handler; + if (this.inputChannel instanceof ExecutorChannelInterceptorAware) { + this.channelInterceptors = ((ExecutorChannelInterceptorAware) this.inputChannel).getChannelInterceptors(); + } + else { + channelInterceptors = null; + } } @@ -59,7 +79,6 @@ public class PollingConsumer extends AbstractPollingEndpoint { super.doStart(); } - @Override protected void doStop() { if (this.handler instanceof Lifecycle) { @@ -68,18 +87,82 @@ public class PollingConsumer extends AbstractPollingEndpoint { super.doStop(); } - @Override protected void handleMessage(Message message) { - this.handler.handleMessage(message); + Deque interceptorStack = null; + try { + if (this.channelInterceptors != null + && ((ExecutorChannelInterceptorAware) this.inputChannel).hasExecutorInterceptors()) { + interceptorStack = new ArrayDeque(); + message = applyBeforeHandle(message, interceptorStack); + if (message == null) { + return; + } + } + this.handler.handleMessage(message); + if (!CollectionUtils.isEmpty(interceptorStack)) { + triggerAfterMessageHandled(message, null, interceptorStack); + } + } + catch (Exception ex) { + if (!CollectionUtils.isEmpty(interceptorStack)) { + triggerAfterMessageHandled(message, ex, interceptorStack); + } + if (ex instanceof MessagingException) { + throw (MessagingException) ex; + } + String description = "Failed to handle " + message + " to " + this + " in " + this.handler; + throw new MessageDeliveryException(message, description, ex); + } + catch (Error ex) {//NOSONAR - ok, we re-throw below + if (!CollectionUtils.isEmpty(interceptorStack)) { + String description = "Failed to handle " + message + " to " + this + " in " + this.handler; + triggerAfterMessageHandled(message, + new MessageDeliveryException(message, description, ex), + interceptorStack); + } + throw ex; + } + } + + private Message applyBeforeHandle(Message message, Deque interceptorStack) { + for (ChannelInterceptor interceptor : this.channelInterceptors) { + if (interceptor instanceof ExecutorChannelInterceptor) { + ExecutorChannelInterceptor executorInterceptor = (ExecutorChannelInterceptor) interceptor; + message = executorInterceptor.beforeHandle(message, this.inputChannel, this.handler); + if (message == null) { + if (logger.isDebugEnabled()) { + logger.debug(executorInterceptor.getClass().getSimpleName() + + " returned null from beforeHandle, i.e. precluding the send."); + } + triggerAfterMessageHandled(null, null, interceptorStack); + return null; + } + interceptorStack.add(executorInterceptor); + } + } + return message; + } + + private void triggerAfterMessageHandled(Message message, Exception ex, + Deque interceptorStack) { + Iterator iterator = interceptorStack.descendingIterator(); + while (iterator.hasNext()) { + ExecutorChannelInterceptor interceptor = iterator.next(); + try { + interceptor.afterMessageHandled(message, this.inputChannel, this.handler, ex); + } + catch (Throwable ex2) {//NOSONAR + logger.error("Exception from afterMessageHandled in " + interceptor, ex2); + } + } } @Override protected Message receiveMessage() { - Message message = (this.receiveTimeout >= 0) + return (this.receiveTimeout >= 0) ? this.inputChannel.receive(this.receiveTimeout) : this.inputChannel.receive(); - return message; } @Override @@ -91,4 +174,5 @@ public class PollingConsumer extends AbstractPollingEndpoint { protected String getResourceKey() { return IntegrationResourceHolder.INPUT_CHANNEL; } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/ExecutorChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/ExecutorChannelTests.java index cf7f1397fb..83d24812f3 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/ExecutorChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/ExecutorChannelTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2015 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,25 +20,43 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; +import static org.mockito.BDDMockito.willThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import org.junit.Before; import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Captor; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; +import org.springframework.beans.factory.BeanFactory; import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.core.task.SyncTaskExecutor; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHandler; import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; +import org.springframework.messaging.support.ChannelInterceptorAdapter; +import org.springframework.messaging.support.ExecutorChannelInterceptor; import org.springframework.messaging.support.GenericMessage; +import org.springframework.messaging.support.MessageBuilder; import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; /** * @author Mark Fisher + * @author Artem Bilan */ public class ExecutorChannelTests { @@ -156,6 +174,50 @@ public class ExecutorChannelTests { assertEquals(numberOfMessages, handler2.count.get()); } + @Test + public void interceptorWithModifiedMessage() { + ExecutorChannel channel = new ExecutorChannel(new SyncTaskExecutor()); + channel.setBeanFactory(mock(BeanFactory.class)); + channel.afterPropertiesSet(); + + MessageHandler handler = mock(MessageHandler.class); + Message expected = mock(Message.class); + BeforeHandleInterceptor interceptor = new BeforeHandleInterceptor(); + interceptor.setMessageToReturn(expected); + channel.addInterceptor(interceptor); + channel.subscribe(handler); + channel.send(new GenericMessage("foo")); + verify(handler).handleMessage(expected); + assertEquals(1, interceptor.getCounter().get()); + assertTrue(interceptor.wasAfterHandledInvoked()); + } + + @Test + public void interceptorWithException() { + ExecutorChannel channel = new ExecutorChannel(new SyncTaskExecutor()); + channel.setBeanFactory(mock(BeanFactory.class)); + channel.afterPropertiesSet(); + + Message message = new GenericMessage("foo"); + + MessageHandler handler = mock(MessageHandler.class); + IllegalStateException expected = new IllegalStateException("Fake exception"); + willThrow(expected).given(handler).handleMessage(message); + BeforeHandleInterceptor interceptor = new BeforeHandleInterceptor(); + channel.addInterceptor(interceptor); + channel.subscribe(handler); + try { + channel.send(message); + } + catch (MessageDeliveryException actual) { + assertSame(expected, actual.getCause()); + } + verify(handler).handleMessage(message); + assertEquals(1, interceptor.getCounter().get()); + assertTrue(interceptor.wasAfterHandledInvoked()); + } + + private static class TestHandler implements MessageHandler { @@ -181,4 +243,40 @@ public class ExecutorChannelTests { } } + private static class BeforeHandleInterceptor extends ChannelInterceptorAdapter + implements ExecutorChannelInterceptor { + + private AtomicInteger counter = new AtomicInteger(); + + private volatile boolean afterHandledInvoked; + + private Message messageToReturn; + + public void setMessageToReturn(Message messageToReturn) { + this.messageToReturn = messageToReturn; + } + + public AtomicInteger getCounter() { + return this.counter; + } + + public boolean wasAfterHandledInvoked() { + return this.afterHandledInvoked; + } + + @Override + public Message beforeHandle(Message message, MessageChannel channel, MessageHandler handler) { + assertNotNull(message); + this.counter.incrementAndGet(); + return (this.messageToReturn != null ? this.messageToReturn : message); + } + + @Override + public void afterMessageHandled(Message message, MessageChannel channel, MessageHandler handler, + Exception ex) { + this.afterHandledInvoked = true; + } + + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java index 62de0377df..9aeedea4c4 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -24,7 +24,10 @@ import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; +import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -36,11 +39,16 @@ import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.channel.ChannelInterceptorAware; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.ChannelInterceptorAdapter; +import org.springframework.messaging.support.ExecutorChannelInterceptor; import org.springframework.messaging.support.GenericMessage; import org.springframework.util.StringUtils; @@ -253,6 +261,45 @@ public class ChannelInterceptorTests { ac.close(); } + @Test + public void testPollingConsumerWithExecutorInterceptor() throws InterruptedException { + TestUtils.TestApplicationContext testApplicationContext = TestUtils.createTestApplicationContext(); + + QueueChannel channel = new QueueChannel(); + + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(2); + final List> messages = new ArrayList<>(); + + PollingConsumer consumer = new PollingConsumer(channel, new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + messages.add(message); + latch1.countDown(); + latch2.countDown(); + } + + }); + + testApplicationContext.registerBean("consumer", consumer); + testApplicationContext.refresh(); + + channel.send(new GenericMessage<>("foo")); + + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + + channel.addInterceptor(new TestExecutorInterceptor()); + channel.send(new GenericMessage<>("foo")); + + assertTrue(latch2.await(10, TimeUnit.SECONDS)); + assertEquals(2, messages.size()); + + assertEquals("foo", messages.get(0).getPayload()); + assertEquals("FOO", messages.get(1).getPayload()); + + testApplicationContext.close(); + } public static class PreSendReturnsMessageInterceptor extends ChannelInterceptorAdapter { private String foo; @@ -374,6 +421,7 @@ public class ChannelInterceptorTests { public void afterReceiveCompletion(Message message, MessageChannel channel, Exception ex) { this.afterCompletionInvoked = true; } + } @@ -386,6 +434,26 @@ public class ChannelInterceptorTests { counter.incrementAndGet(); return false; } + } + private static class TestExecutorInterceptor extends ChannelInterceptorAdapter + implements ExecutorChannelInterceptor { + + @Override + public Message beforeHandle(Message message, MessageChannel channel, MessageHandler handler) { + return MessageBuilder.withPayload(((String) message.getPayload()).toUpperCase()) + .copyHeaders(message.getHeaders()) + .build(); + } + + @Override + public void afterMessageHandled(Message message, MessageChannel channel, MessageHandler handler, + Exception ex) { + + } + + } + + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests-context.xml index c267f01b79..f775ec1970 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests-context.xml @@ -11,6 +11,8 @@ + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests.java index 44b73eff3f..4e856590d5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -34,6 +34,7 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -41,10 +42,11 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * @author Dave Syer * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) -@MessageEndpoint +@DirtiesContext public class AnnotatedEndpointActivationTests { @Autowired @@ -63,21 +65,6 @@ public class AnnotatedEndpointActivationTests { // them will get the message. private static volatile int count = 0; - - @ServiceActivator(inputChannel = "input", outputChannel = "output") - public String process(String message) { - count++; - String result = message + ": " + count; - return result; - } - - @ServiceActivator(inputChannel = "inputImplicit", outputChannel = "output") - public String processImplicit(String message) { - count++; - String result = message + ": " + count; - return result; - } - @Before public void resetCount() { count = 0; @@ -108,12 +95,14 @@ public class AnnotatedEndpointActivationTests { } @Test(expected = MessageDeliveryException.class) + @DirtiesContext public void stopContext() { applicationContext.stop(); this.input.send(new GenericMessage("foo")); } @Test + @DirtiesContext public void stopAndRestartContext() { applicationContext.stop(); applicationContext.start(); @@ -124,4 +113,21 @@ public class AnnotatedEndpointActivationTests { assertEquals(1, count); } + @MessageEndpoint + private static class AnnotatedEndpoint { + + @ServiceActivator(inputChannel = "input", outputChannel = "output") + public String process(String message) { + count++; + return message + ": " + count; + } + + @ServiceActivator(inputChannel = "inputImplicit", outputChannel = "output") + public String processImplicit(String message) { + count++; + return message + ": " + count; + } + + } + } diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-4.2.xsd b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-4.2.xsd index 594b494349..b40ae59fad 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-4.2.xsd +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-4.2.xsd @@ -1255,7 +1255,7 @@ interceptors) { + super.setInterceptors(interceptors); + for (ChannelInterceptor interceptor : interceptors) { + if (interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize++; + } + } + } + + @Override + public void addInterceptor(ChannelInterceptor interceptor) { + super.addInterceptor(interceptor); + if (interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize++; + } + } + + @Override + public void addInterceptor(int index, ChannelInterceptor interceptor) { + super.addInterceptor(index, interceptor); + if (interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize++; + } + } + + @Override + public boolean removeInterceptor(ChannelInterceptor interceptor) { + boolean removed = super.removeInterceptor(interceptor); + if (removed && interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize--; + } + return removed; + } + + @Override + public ChannelInterceptor removeInterceptor(int index) { + ChannelInterceptor interceptor = super.removeInterceptor(index); + if (interceptor != null && interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize--; + } + return interceptor; + } + + @Override + public boolean hasExecutorInterceptors() { + return this.executorInterceptorsSize > 0; + } + } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCachedConsumersTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCachedConsumersTests.java index 28801fb16c..28325170e0 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCachedConsumersTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCachedConsumersTests.java @@ -37,7 +37,6 @@ import org.springframework.integration.gateway.RequestReplyExchanger; import org.springframework.integration.jms.ActiveMQMultiContextTests; import org.springframework.integration.jms.JmsOutboundGateway; import org.springframework.integration.jms.config.ActiveMqTestUtils; -import org.springframework.messaging.support.GenericMessage; import org.springframework.integration.test.support.LongRunningIntegrationTest; import org.springframework.integration.test.util.TestUtils; import org.springframework.jms.connection.CachingConnectionFactory; @@ -46,6 +45,7 @@ import org.springframework.jms.core.MessageCreator; import org.springframework.jms.listener.DefaultMessageListenerContainer; import org.springframework.jms.listener.SessionAwareMessageListener; import org.springframework.jms.support.converter.SimpleMessageConverter; +import org.springframework.messaging.support.GenericMessage; /** * @author Oleg Zhurakousky * @author Gary Russell @@ -213,7 +213,6 @@ public class RequestReplyScenariosWithCachedConsumersTests extends ActiveMQMulti final Destination replyDestination = context.getBean("siInQueueE", Destination.class); for (int i = 0; i < 3; i++) { - System.out.println("#### " + i); try { gateway.exchange(gateway.exchange(new GenericMessage("foo"))); } catch (Exception e) {/*ignore*/} diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/MBeanAttributeFilterTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/MBeanAttributeFilterTests.java index 81e444a1ea..aaee07c622 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/MBeanAttributeFilterTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/MBeanAttributeFilterTests.java @@ -111,7 +111,6 @@ public class MBeanAttributeFilterTests { List keys = new ArrayList(bean.keySet()); Collections.sort(keys); - System.out.println(keys); assertThat(keys, contains("LoggingEnabled", "MaxSendDuration", "MeanErrorRate", diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/NotificationPublishingChannelAdapterParserTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/NotificationPublishingChannelAdapterParserTests.java index 72e8b3c650..3de06f9377 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/NotificationPublishingChannelAdapterParserTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/NotificationPublishingChannelAdapterParserTests.java @@ -26,7 +26,6 @@ import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import java.util.Set; - import javax.management.Attribute; import javax.management.MBeanException; import javax.management.MBeanServer; @@ -179,7 +178,6 @@ public class NotificationPublishingChannelAdapterParserTests { @Override protected Object doInvoke(ExecutionCallback callback, Object target, Message message) throws Exception { adviceCalled++; - System.out.println("foo"); new RuntimeException("foo").printStackTrace(); return callback.execute(); } diff --git a/spring-integration-security/src/main/java/org/springframework/integration/security/channel/ChannelAccessPolicy.java b/spring-integration-security/src/main/java/org/springframework/integration/security/channel/ChannelAccessPolicy.java index 6299d06d8d..c7bf20f802 100644 --- a/spring-integration-security/src/main/java/org/springframework/integration/security/channel/ChannelAccessPolicy.java +++ b/spring-integration-security/src/main/java/org/springframework/integration/security/channel/ChannelAccessPolicy.java @@ -23,7 +23,7 @@ import org.springframework.security.access.ConfigAttribute; /** * Interface to encapsulate {@link ConfigAttribute}s for secured channel * send and receive operations. - * + * * @author Oleg Zhurakousky * @since 2.0 */ diff --git a/spring-integration-security/src/main/java/org/springframework/integration/security/channel/ChannelSecurityInterceptor.java b/spring-integration-security/src/main/java/org/springframework/integration/security/channel/ChannelSecurityInterceptor.java index a82fd17bb1..5c3dbdbef4 100644 --- a/spring-integration-security/src/main/java/org/springframework/integration/security/channel/ChannelSecurityInterceptor.java +++ b/spring-integration-security/src/main/java/org/springframework/integration/security/channel/ChannelSecurityInterceptor.java @@ -20,6 +20,7 @@ import java.lang.reflect.Method; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; + import org.springframework.security.access.SecurityMetadataSource; import org.springframework.security.access.intercept.AbstractSecurityInterceptor; import org.springframework.security.access.intercept.InterceptorStatusToken; diff --git a/spring-integration-security/src/main/java/org/springframework/integration/security/channel/SecurityContextPropagationChannelInterceptor.java b/spring-integration-security/src/main/java/org/springframework/integration/security/channel/SecurityContextPropagationChannelInterceptor.java new file mode 100644 index 0000000000..eded5c1dda --- /dev/null +++ b/spring-integration-security/src/main/java/org/springframework/integration/security/channel/SecurityContextPropagationChannelInterceptor.java @@ -0,0 +1,95 @@ +/* + * Copyright 2015 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.security.channel; + +import org.springframework.aop.support.AopUtils; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.interceptor.ThreadStatePropagationChannelInterceptor; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.support.ExecutorChannelInterceptor; +import org.springframework.security.core.Authentication; +import org.springframework.security.core.context.SecurityContext; +import org.springframework.security.core.context.SecurityContextHolder; + +/** + * The {@link ExecutorChannelInterceptor} implementation responsible for + * the {@link SecurityContext} propagation from one message flow's thread to another + * through the {@link MessageChannel}s involved in the flow. + *

+ * In addition this interceptor cleans up (restores) the {@link SecurityContext} + * in the containers Threads for channels like + * {@link org.springframework.integration.channel.ExecutorChannel} + * and {@link org.springframework.integration.channel.QueueChannel}. + * + * @author Artem Bilan + * @see ThreadStatePropagationChannelInterceptor + * @since 4.2 + */ +public class SecurityContextPropagationChannelInterceptor + extends ThreadStatePropagationChannelInterceptor { + + private final static SecurityContext EMPTY_CONTEXT = SecurityContextHolder.createEmptyContext(); + + private static final ThreadLocal ORIGINAL_CONTEXT = new ThreadLocal(); + + @Override + public void afterMessageHandled(Message message, MessageChannel channel, MessageHandler handler, Exception ex) { + cleanup(); + } + + @Override + protected Authentication obtainPropagatingContext(Message message, MessageChannel channel) { + if (!DirectChannel.class.isAssignableFrom(AopUtils.getTargetClass(channel))) { + return SecurityContextHolder.getContext().getAuthentication(); + } + return null; + } + + @Override + protected void populatePropagatedContext(Authentication authentication, Message message, + MessageChannel channel) { + if (authentication != null) { + SecurityContext currentContext = SecurityContextHolder.getContext(); + + ORIGINAL_CONTEXT.set(currentContext); + + SecurityContext context = SecurityContextHolder.createEmptyContext(); + context.setAuthentication(authentication); + SecurityContextHolder.setContext(context); + } + } + + public static void cleanup() { + SecurityContext originalContext = ORIGINAL_CONTEXT.get(); + + try { + if (originalContext == null || EMPTY_CONTEXT.equals(originalContext)) { + SecurityContextHolder.clearContext(); + ORIGINAL_CONTEXT.remove(); + } + else { + SecurityContextHolder.setContext(originalContext); + } + } + catch (Throwable t) { + SecurityContextHolder.clearContext(); + } + } + +} diff --git a/spring-integration-security/src/test/java/org/springframework/integration/security/channel/ChannelAdapterSecurityIntegrationTests-context.xml b/spring-integration-security/src/test/java/org/springframework/integration/security/channel/ChannelAdapterSecurityIntegrationTests-context.xml index de11961211..cace900e11 100644 --- a/spring-integration-security/src/test/java/org/springframework/integration/security/channel/ChannelAdapterSecurityIntegrationTests-context.xml +++ b/spring-integration-security/src/test/java/org/springframework/integration/security/channel/ChannelAdapterSecurityIntegrationTests-context.xml @@ -1,14 +1,16 @@ + http://www.springframework.org/schema/integration/security/spring-integration-security.xsd + http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd"> @@ -24,4 +26,27 @@ + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-security/src/test/java/org/springframework/integration/security/channel/ChannelAdapterSecurityIntegrationTests.java b/spring-integration-security/src/test/java/org/springframework/integration/security/channel/ChannelAdapterSecurityIntegrationTests.java index eb8d9647a4..93b45ae271 100644 --- a/spring-integration-security/src/test/java/org/springframework/integration/security/channel/ChannelAdapterSecurityIntegrationTests.java +++ b/spring-integration-security/src/test/java/org/springframework/integration/security/channel/ChannelAdapterSecurityIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -16,24 +16,32 @@ package org.springframework.integration.security.channel; +import static org.hamcrest.Matchers.instanceOf; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; import org.junit.After; import org.junit.Test; +import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.integration.security.TestHandler; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.support.GenericMessage; import org.springframework.integration.security.SecurityTestUtils; +import org.springframework.integration.security.TestHandler; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.GenericMessage; import org.springframework.security.access.AccessDeniedException; +import org.springframework.security.authentication.AuthenticationCredentialsNotFoundException; import org.springframework.security.core.AuthenticationException; import org.springframework.security.core.context.SecurityContext; import org.springframework.security.core.context.SecurityContextHolder; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Mark Fisher @@ -41,7 +49,9 @@ import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests; * @author Artem Bilan */ @ContextConfiguration -public class ChannelAdapterSecurityIntegrationTests extends AbstractJUnit4SpringContextTests { +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD) +public class ChannelAdapterSecurityIntegrationTests { @Autowired @Qualifier("securedChannelAdapter") @@ -55,6 +65,18 @@ public class ChannelAdapterSecurityIntegrationTests extends AbstractJUnit4Spring @Qualifier("unsecuredChannelAdapter") MessageChannel unsecuredChannelAdapter; + @Autowired + @Qualifier("queueChannel") + MessageChannel queueChannel; + + @Autowired + @Qualifier("securedChannelQueue") + PollableChannel securedChannelQueue; + + @Autowired + @Qualifier("errorChannel") + PollableChannel errorChannel; + @Autowired TestHandler testConsumer; @@ -66,14 +88,12 @@ public class ChannelAdapterSecurityIntegrationTests extends AbstractJUnit4Spring @Test(expected = AccessDeniedException.class) - @DirtiesContext public void testSecuredWithNotEnoughPermission() { login("bob", "bobspassword", "ROLE_ADMINA"); securedChannelAdapter.send(new GenericMessage("test")); } @Test - @DirtiesContext public void testSecuredWithPermission() { login("bob", "bobspassword", "ROLE_ADMIN", "ROLE_PRESIDENT"); securedChannelAdapter.send(new GenericMessage("test")); @@ -81,28 +101,42 @@ public class ChannelAdapterSecurityIntegrationTests extends AbstractJUnit4Spring assertEquals("Wrong size of message list in target", 2, testConsumer.sentMessages.size()); } + @Test + public void testSecurityContextPropagation() { + login("bob", "bobspassword", "ROLE_ADMIN", "ROLE_PRESIDENT"); + this.queueChannel.send(new GenericMessage("test")); + Message receive = this.securedChannelQueue.receive(10000); + assertNotNull(receive); + + SecurityContextHolder.clearContext(); + + this.queueChannel.send(new GenericMessage("test")); + Message errorMessage = this.errorChannel.receive(1000); + assertNotNull(errorMessage); + Object payload = errorMessage.getPayload(); + assertThat(payload, instanceOf(MessageHandlingException.class)); + assertThat(((MessageHandlingException) payload).getCause(), + instanceOf(AuthenticationCredentialsNotFoundException.class)); + } + @Test(expected = AccessDeniedException.class) - @DirtiesContext - public void testSecuredWithoutPermision() { + public void testSecuredWithoutPermission() { login("bob", "bobspassword", "ROLE_USER"); securedChannelAdapter.send(new GenericMessage("test")); } @Test(expected = AccessDeniedException.class) - @DirtiesContext - public void testSecured2WithoutPermision() { + public void testSecured2WithoutPermission() { login("bob", "bobspassword", "ROLE_USER"); securedChannelAdapter2.send(new GenericMessage("test")); } @Test(expected = AuthenticationException.class) - @DirtiesContext public void testSecuredWithoutAuthenticating() { securedChannelAdapter.send(new GenericMessage("test")); } @Test - @DirtiesContext public void testUnsecuredAsAdmin() { login("bob", "bobspassword", "ROLE_ADMIN"); unsecuredChannelAdapter.send(new GenericMessage("test")); @@ -110,7 +144,6 @@ public class ChannelAdapterSecurityIntegrationTests extends AbstractJUnit4Spring } @Test - @DirtiesContext public void testUnsecuredAsUser() { login("bob", "bobspassword", "ROLE_USER"); unsecuredChannelAdapter.send(new GenericMessage("test")); @@ -118,7 +151,6 @@ public class ChannelAdapterSecurityIntegrationTests extends AbstractJUnit4Spring } @Test - @DirtiesContext public void testUnsecuredWithoutAuthenticating() { unsecuredChannelAdapter.send(new GenericMessage("test")); assertEquals("Wrong size of message list in target", 1, testConsumer.sentMessages.size()); diff --git a/spring-integration-security/src/test/java/org/springframework/integration/security/channel/ChannelSecurityInterceptorTests.java b/spring-integration-security/src/test/java/org/springframework/integration/security/channel/ChannelSecurityInterceptorTests.java index d764a63766..d9af6b26ad 100644 --- a/spring-integration-security/src/test/java/org/springframework/integration/security/channel/ChannelSecurityInterceptorTests.java +++ b/spring-integration-security/src/test/java/org/springframework/integration/security/channel/ChannelSecurityInterceptorTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2015 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. diff --git a/spring-integration-security/src/test/java/org/springframework/integration/security/config/ChannelSecurityInterceptorSecuredChannelAnnotationTests.java b/spring-integration-security/src/test/java/org/springframework/integration/security/config/ChannelSecurityInterceptorSecuredChannelAnnotationTests.java index f2b4ee41df..a5b06b25ff 100644 --- a/spring-integration-security/src/test/java/org/springframework/integration/security/config/ChannelSecurityInterceptorSecuredChannelAnnotationTests.java +++ b/spring-integration-security/src/test/java/org/springframework/integration/security/config/ChannelSecurityInterceptorSecuredChannelAnnotationTests.java @@ -16,27 +16,56 @@ package org.springframework.integration.security.config; +import static org.hamcrest.Matchers.instanceOf; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; + +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.Executors; import org.junit.After; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.ImportResource; +import org.springframework.integration.annotation.BridgeTo; +import org.springframework.integration.annotation.Poller; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.channel.ChannelInterceptorAware; import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.ExecutorChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.config.GlobalChannelInterceptor; +import org.springframework.integration.router.RecipientListRouter; import org.springframework.integration.security.SecurityTestUtils; import org.springframework.integration.security.TestHandler; import org.springframework.integration.security.channel.ChannelSecurityInterceptor; import org.springframework.integration.security.channel.SecuredChannel; +import org.springframework.integration.security.channel.SecurityContextPropagationChannelInterceptor; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.PollableChannel; import org.springframework.messaging.SubscribableChannel; +import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.GenericMessage; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.security.access.AccessDecisionManager; import org.springframework.security.access.AccessDeniedException; +import org.springframework.security.authentication.AuthenticationCredentialsNotFoundException; import org.springframework.security.authentication.AuthenticationManager; import org.springframework.security.core.AuthenticationException; import org.springframework.security.core.context.SecurityContext; @@ -63,6 +92,22 @@ public class ChannelSecurityInterceptorSecuredChannelAnnotationTests { @Autowired MessageChannel unsecuredChannel; + @Autowired + @Qualifier("queueChannel") + MessageChannel queueChannel; + + @Autowired + @Qualifier("securedChannelQueue") + PollableChannel securedChannelQueue; + + @Autowired + @Qualifier("executorChannel") + MessageChannel executorChannel; + + @Autowired + @Qualifier("errorChannel") + PollableChannel errorChannel; + @Autowired TestHandler testConsumer; @@ -124,6 +169,42 @@ public class ChannelSecurityInterceptorSecuredChannelAnnotationTests { assertEquals("Wrong size of message list in target", 1, testConsumer.sentMessages.size()); } + @Test + public void testSecurityContextPropagationQueueChannel() { + login("bob", "bobspassword", "ROLE_ADMIN", "ROLE_PRESIDENT"); + this.queueChannel.send(new GenericMessage("test")); + Message receive = this.securedChannelQueue.receive(10000); + assertNotNull(receive); + + SecurityContextHolder.clearContext(); + + this.queueChannel.send(new GenericMessage("test")); + Message errorMessage = this.errorChannel.receive(1000); + assertNotNull(errorMessage); + Object payload = errorMessage.getPayload(); + assertThat(payload, instanceOf(MessageHandlingException.class)); + assertThat(((MessageHandlingException) payload).getCause(), + instanceOf(AuthenticationCredentialsNotFoundException.class)); + } + + @Test + public void testSecurityContextPropagationExecutorChannel() { + login("bob", "bobspassword", "ROLE_ADMIN", "ROLE_PRESIDENT"); + this.executorChannel.send(new GenericMessage("test")); + Message receive = this.securedChannelQueue.receive(10000); + assertNotNull(receive); + + SecurityContextHolder.clearContext(); + + this.queueChannel.send(new GenericMessage("test")); + Message errorMessage = this.errorChannel.receive(1000); + assertNotNull(errorMessage); + Object payload = errorMessage.getPayload(); + assertThat(payload, instanceOf(MessageHandlingException.class)); + assertThat(((MessageHandlingException) payload).getCause(), + instanceOf(AuthenticationCredentialsNotFoundException.class)); + } + private void login(String username, String password, String... roles) { SecurityContext context = SecurityTestUtils.createContext(username, password, roles); @@ -153,6 +234,41 @@ public class ChannelSecurityInterceptorSecuredChannelAnnotationTests { return new DirectChannel(); } + @Bean + @GlobalChannelInterceptor(patterns = {"queueChannel", "executorChannel"}) + public ChannelInterceptor securityContextPropagationInterceptor() { + return new SecurityContextPropagationChannelInterceptor(); + } + + @Bean + @BridgeTo(value = "securedChannelQueue", poller = @Poller(fixedDelay = "1000")) + public PollableChannel queueChannel() { + return new QueueChannel(); + } + + @Bean + @SecuredChannel(interceptor = "channelSecurityInterceptor", sendAccess = {"ROLE_ADMIN", "ROLE_PRESIDENT"}) + public PollableChannel securedChannelQueue() { + return new QueueChannel(); + } + + @Bean + @BridgeTo("securedChannelQueue") + public SubscribableChannel executorChannel() { + return new ExecutorChannel(Executors.newSingleThreadExecutor()); + } + + + @Bean + public TaskScheduler taskScheduler() { + return new ThreadPoolTaskScheduler(); + } + + @Bean + public PollableChannel errorChannel() { + return new QueueChannel(); + } + @Bean public TestHandler testHandler() { TestHandler testHandler = new TestHandler(); diff --git a/spring-integration-security/src/test/java/org/springframework/integration/security/config/SecuredChannelsParserTests.java b/spring-integration-security/src/test/java/org/springframework/integration/security/config/SecuredChannelsParserTests.java index 5cac7c605b..21e93f81af 100644 --- a/spring-integration-security/src/test/java/org/springframework/integration/security/config/SecuredChannelsParserTests.java +++ b/spring-integration-security/src/test/java/org/springframework/integration/security/config/SecuredChannelsParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2015 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. diff --git a/spring-integration-security/src/test/resources/log4j.properties b/spring-integration-security/src/test/resources/log4j.properties index 857ef41ba2..84f5f1959d 100644 --- a/spring-integration-security/src/test/resources/log4j.properties +++ b/spring-integration-security/src/test/resources/log4j.properties @@ -2,7 +2,7 @@ log4j.rootCategory=WARN, stdout log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout -log4j.appender.stdout.layout.ConversionPattern=%c{1}: %m%n +log4j.appender.stdout.layout.ConversionPattern=%d %5p %c{1} [%t] : %m%n log4j.category.org.springframework.integration.security=WARN log4j.category.org.springframework.integration=WARN diff --git a/spring-integration-ws/src/test/java/org/springframework/integration/ws/config/WebServiceInboundGatewayParserTests.java b/spring-integration-ws/src/test/java/org/springframework/integration/ws/config/WebServiceInboundGatewayParserTests.java index 52f75f598b..70dad46e9e 100644 --- a/spring-integration-ws/src/test/java/org/springframework/integration/ws/config/WebServiceInboundGatewayParserTests.java +++ b/spring-integration-ws/src/test/java/org/springframework/integration/ws/config/WebServiceInboundGatewayParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -179,7 +179,6 @@ public class WebServiceInboundGatewayParserTests { MessageHistory history = MessageHistory.read(message); assertNotNull(history); Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, "extractsPayload", 0); - System.out.println(componentHistoryRecord); assertNotNull(componentHistoryRecord); assertEquals("ws:inbound-gateway", componentHistoryRecord.get("type")); } diff --git a/src/reference/asciidoc/security.adoc b/src/reference/asciidoc/security.adoc index 73ac688d86..a04eb13ca1 100644 --- a/src/reference/asciidoc/security.adoc +++ b/src/reference/asciidoc/security.adoc @@ -4,7 +4,16 @@ [[security-intro]] === Introduction -Spring Integration builds upon the http://static.springframework.org/spring-security/site/[Spring Security project] to enable role based security checks to be applied to channel send and receive invocations. +Security is one of the important functions in any modern enterprise (or cloud) application, +moreover it is critical for distributed systems, such as those built using Enterprise +Integration Patterns. +Messaging independence and loosely-coupling allow target systems to communicate with each other +with any type of data in the message's `payload`. +We can either trust all those messages or _secure_ our service against "infecting" messages. + +Spring Integration together with +http://projects.spring.io/spring-security/[Spring Security] provide a simple and comprehensive way to +secure message channels, as well as other part of the integration solution. [[securing-channels]] === Securing channels @@ -70,7 +79,7 @@ With the `@SecuredChannel` annotation, the Java configuration variant of the XML @EnableIntegration public class ContextConfiguration { - @Bean + @Bean @SecuredChannel(interceptor = "channelSecurityInterceptor", sendAccess = "ROLE_ADMIN") public SubscribableChannel adminChannel() { return new DirectChannel(); @@ -93,3 +102,60 @@ public class ContextConfiguration { } ---- + +[[security-context-propagation]] +=== SecurityContext Propagation + +To be sure that our interaction with the application is secure, according to its security system rules, we should supply +some _security context_ with an _authentication_ (principal) object. +The Spring Security project provides a flexible, canonical mechanism to authenticate our application clients +over HTTP, WebSocket or SOAP protocols (as can be done for any other integration protocol +with a simple Spring Security extension) and it provides a `SecurityContext` for further authorization checks on the +application objects, such as message channels. +By default, the `SecurityContext` is tied with the current `Thread` 's execution state using the +(`ThreadLocalSecurityContextHolderStrategy`). +It is accessed by an AOP interceptor on secured methods to check if that `principal` of the invocation has +sufficent permissions to call that method, for example. +This works well with the current thread, but often, processing logic can be performed on another thread or even +on several threads, or on to some external system(s). + +Standard thread-bound behavior is easy to configure if our application is built on the Spring Integration components +and its message channels. +In this case, the secured objects may be any service activator or transformer, secured with a +`MethodSecurityInterceptor` in their `` (see <>) +or even `MessageChannel` (see <> above). +When using `DirectChannel` communication, the `SecurityContext` is available +automatically, because the downstream flow runs on the current thread. +But in case of the `QueueChannel`, `ExecutorChannel` and `PublishSubscribeChannel` with an `Executor`, messages are +transferred from one thread to another (or several) by the nature of those channels. +In order to support such scenarios, +we can either transfer an `Authentication` object within the message headers and extract and authenticate it on the +other side before secured object access. +Or, we can _propagate_ the `SecurityContext` to the thread receiving the transferred message. + +Starting with _version 4.2_ `SecurityContext` propagation has been introduced. +It is implemented as a `SecurityContextPropagationChannelInterceptor`, which can simply be added to any `MessageChannel` +or configured as a `@GlobalChannelInterceptor`. +The logic of this interceptor is based on the `SecurityContext` extraction from the current thread from the `preSend()` +method, and its populating to another thread from the `postReceive()` (`beforeHandle()`) method. +Actually, this interceptor is an extension of the more generic `ThreadStatePropagationChannelInterceptor`, which wraps +the message-to-send together with the state-to-propagate in an internal `Message` extension - +`MessageWithThreadState`, - on one side and extracts the original message back and state-to-propagate on another. +The `ThreadStatePropagationChannelInterceptor` can be extended for any context propagation use-case and +`SecurityContextPropagationChannelInterceptor` is a good sample on the matter. + +IMPORTANT: Since the logic of the `ThreadStatePropagationChannelInterceptor` is based on message modification +(it returns an internal `MessageWithThreadState` object to send), you should be careful when combining this +interceptor with any other which is intended to modify messages too, e.g. through the +`MessageBuilder.withPayload(...)...build()` - the state-to-propagate may be lost. +In most cases to overcome the issue, it's sufficient to order interceptors for the channel and ensure the + `ThreadStatePropagationChannelInterceptor` is the last one in the stack. + +Propagation and population of `SecurityContext` is just one half of the work. +Since the message isn't an owner of the threads in the message flow and we should be sure that we are secure against +any incoming messages, we have to _clean up_ the `SecurityContext` from `ThreadLocal`. +The `SecurityContextPropagationChannelInterceptor` provides `afterMessageHandled()` interceptor's method +implementation to do the clean up operation to free the Thread in the end of invocation from that propagated principal. +This means that, when the thread that processes the handed-off message, completes the processing of the message +(successfully or otherwise), the context is cleared so that it can't be inadvertently be used when processing another +message. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 126a8436e3..e2448c9d50 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -19,7 +19,7 @@ However, this has some important implications for (some) user environments. For complete details, see <> and <>. [[x4.2-mongodb-metadata-store]] -==== MongodDB Metadata Store +==== MongoDB Metadata Store The `MongoDbMetadataStore` is now available. For more information, see <>. @@ -29,6 +29,13 @@ The `MongoDbMetadataStore` is now available. For more information, see <>. +[[x4.2-security-context-propagation]] +==== SecurityContext Propagation + +The `SecurityContextPropagationChannelInterceptor` has been +introduced for the `SecurityContext` propagation from one message flow's Thread to another. +For more information, see <>. + [[x4.2-file-splitter]] ==== FileSplitter