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 index 01e489b124..7d5d5212fe 100644 --- 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 @@ -16,6 +16,7 @@ package org.springframework.integration.channel.interceptor; +import org.springframework.integration.support.MessageDecorator; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -91,14 +92,15 @@ public abstract class ThreadStatePropagationChannelInterceptor protected abstract void populatePropagatedContext(S state, Message message, MessageChannel channel); - private static final class MessageWithThreadState implements Message { + private static final class MessageWithThreadState implements Message, MessageDecorator { - private final Message message; + private final Message message; private final S state; + @SuppressWarnings("unchecked") private MessageWithThreadState(Message message, S state) { - this.message = message; + this.message = (Message) message; this.state = state; } @@ -112,6 +114,11 @@ public abstract class ThreadStatePropagationChannelInterceptor return this.message.getHeaders(); } + @Override + public Message decorateMessage(Message message) { + return new MessageWithThreadState(message, this.state); + } + @Override public String toString() { return "MessageWithThreadState{" + 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 d53516c13c..9f2f35c923 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 @@ -17,6 +17,7 @@ package org.springframework.integration.dispatcher; import java.util.Collection; +import java.util.UUID; import java.util.concurrent.Executor; import org.springframework.beans.BeansException; @@ -25,6 +26,7 @@ import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.integration.MessageDispatchingException; import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilderFactory; +import org.springframework.integration.support.MessageDecorator; import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; @@ -158,13 +160,20 @@ public class BroadcastingDispatcher extends AbstractDispatcher implements BeanFa throw new MessageDispatchingException(message, "Dispatcher has no subscribers"); } int sequenceSize = handlers.size(); + Message messageToSend = message; + UUID sequenceId = null; + if (this.applySequence) { + sequenceId = message.getHeaders().getId(); + } for (MessageHandler handler : handlers) { - Message messageToSend = message; if (this.applySequence) { messageToSend = getMessageBuilderFactory() .fromMessage(message) - .pushSequenceDetails(message.getHeaders().getId(), sequenceNumber++, sequenceSize) + .pushSequenceDetails(sequenceId, sequenceNumber++, sequenceSize) .build(); + if (message instanceof MessageDecorator) { + messageToSend = ((MessageDecorator) message).decorateMessage(messageToSend); + } } if (this.executor != null) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/MessageDecorator.java b/spring-integration-core/src/main/java/org/springframework/integration/support/MessageDecorator.java new file mode 100644 index 0000000000..6306839f60 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/MessageDecorator.java @@ -0,0 +1,34 @@ +/* + * Copyright 2016 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.support; + +import org.springframework.messaging.Message; + +/** + * The {@link Message} decoration contract. + * An implementation may decide to return any {@link Message} instance + * and even a different {@link Message} implementation. Usually used to + * wrap a message in another. + * + * @author Artem Bilan + * @since 4.2.9 + */ +public interface MessageDecorator { + + Message decorateMessage(Message message); + +} 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 30b1dbaf06..c522ac680d 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 @@ -33,13 +33,17 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.ImportResource; import org.springframework.context.support.PropertySourcesPlaceholderConfigurer; +import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.annotation.BridgeTo; import org.springframework.integration.annotation.Poller; +import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.ExecutorChannel; +import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.config.GlobalChannelInterceptor; +import org.springframework.integration.handler.BridgeHandler; import org.springframework.integration.security.SecurityTestUtils; import org.springframework.integration.security.TestHandler; import org.springframework.integration.security.channel.ChannelSecurityInterceptor; @@ -47,6 +51,7 @@ import org.springframework.integration.security.channel.SecuredChannel; import org.springframework.integration.security.channel.SecurityContextPropagationChannelInterceptor; 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; @@ -95,6 +100,14 @@ public class ChannelSecurityInterceptorSecuredChannelAnnotationTests { @Qualifier("executorChannel") MessageChannel executorChannel; + @Autowired + @Qualifier("publishSubscribeChannel") + PublishSubscribeChannel publishSubscribeChannel; + + @Autowired + @Qualifier("securedChannelQueue2") + PollableChannel securedChannelQueue2; + @Autowired @Qualifier("errorChannel") PollableChannel errorChannel; @@ -170,7 +183,7 @@ public class ChannelSecurityInterceptorSecuredChannelAnnotationTests { SecurityContextHolder.clearContext(); this.queueChannel.send(new GenericMessage("test")); - Message errorMessage = this.errorChannel.receive(1000); + Message errorMessage = this.errorChannel.receive(10000); assertNotNull(errorMessage); Object payload = errorMessage.getPayload(); assertThat(payload, instanceOf(MessageHandlingException.class)); @@ -187,8 +200,8 @@ public class ChannelSecurityInterceptorSecuredChannelAnnotationTests { SecurityContextHolder.clearContext(); - this.queueChannel.send(new GenericMessage("test")); - Message errorMessage = this.errorChannel.receive(1000); + this.executorChannel.send(new GenericMessage("test")); + Message errorMessage = this.errorChannel.receive(10000); assertNotNull(errorMessage); Object payload = errorMessage.getPayload(); assertThat(payload, instanceOf(MessageHandlingException.class)); @@ -196,6 +209,48 @@ public class ChannelSecurityInterceptorSecuredChannelAnnotationTests { instanceOf(AuthenticationCredentialsNotFoundException.class)); } + @Test + public void testSecurityContextPropagationPublishSubscribeChannel() { + login("bob", "bobspassword", "ROLE_ADMIN", "ROLE_PRESIDENT"); + + this.publishSubscribeChannel.send(new GenericMessage("test")); + + Message receive = this.securedChannelQueue.receive(10000); + assertNotNull(receive); + IntegrationMessageHeaderAccessor headerAccessor = new IntegrationMessageHeaderAccessor(receive); + assertEquals(new Integer(0), headerAccessor.getSequenceNumber()); + + receive = this.securedChannelQueue2.receive(10000); + assertNotNull(receive); + headerAccessor = new IntegrationMessageHeaderAccessor(receive); + assertEquals(new Integer(0), headerAccessor.getSequenceNumber()); + + this.publishSubscribeChannel.setApplySequence(true); + + this.publishSubscribeChannel.send(new GenericMessage("test")); + + receive = this.securedChannelQueue.receive(10000); + assertNotNull(receive); + headerAccessor = new IntegrationMessageHeaderAccessor(receive); + assertEquals(new Integer(1), headerAccessor.getSequenceNumber()); + + receive = this.securedChannelQueue2.receive(10000); + assertNotNull(receive); + headerAccessor = new IntegrationMessageHeaderAccessor(receive); + assertEquals(new Integer(2), headerAccessor.getSequenceNumber()); + + this.publishSubscribeChannel.setApplySequence(false); + + SecurityContextHolder.clearContext(); + + this.publishSubscribeChannel.send(new GenericMessage("test")); + Message errorMessage = this.errorChannel.receive(10000); + 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); @@ -231,7 +286,10 @@ public class ChannelSecurityInterceptorSecuredChannelAnnotationTests { } @Bean - @GlobalChannelInterceptor(patterns = {"#{'queueChannel'}", "${security.channel:executorChannel}"}) + @GlobalChannelInterceptor(patterns = { + "#{'queueChannel'}", + "${security.channel:executorChannel}", + "publishSubscribeChannel" }) public ChannelInterceptor securityContextPropagationInterceptor() { return new SecurityContextPropagationChannelInterceptor(); } @@ -255,6 +313,35 @@ public class ChannelSecurityInterceptorSecuredChannelAnnotationTests { } + @Bean + public PublishSubscribeChannel publishSubscribeChannel() { + return new PublishSubscribeChannel(Executors.newCachedThreadPool()); + } + + @Bean + @ServiceActivator(inputChannel = "publishSubscribeChannel") + public MessageHandler securedChannelQueueBridge() { + BridgeHandler handler = new BridgeHandler(); + handler.setOutputChannel(securedChannelQueue()); + handler.setOrder(1); + return handler; + } + + @Bean + @SecuredChannel(interceptor = "channelSecurityInterceptor", sendAccess = {"ROLE_ADMIN", "ROLE_PRESIDENT"}) + public PollableChannel securedChannelQueue2() { + return new QueueChannel(); + } + + @Bean + @ServiceActivator(inputChannel = "publishSubscribeChannel") + public MessageHandler securedChannelQueue2Bridge() { + BridgeHandler handler = new BridgeHandler(); + handler.setOutputChannel(securedChannelQueue2()); + handler.setOrder(2); + return handler; + } + @Bean public TaskScheduler taskScheduler() { return new ThreadPoolTaskScheduler();