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 c522ac680d..f66dcbec3a 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 @@ -22,6 +22,8 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import org.junit.After; import org.junit.Test; @@ -33,8 +35,13 @@ 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.core.task.AsyncTaskExecutor; +import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.annotation.BridgeTo; +import org.springframework.integration.annotation.Gateway; +import org.springframework.integration.annotation.IntegrationComponentScan; +import org.springframework.integration.annotation.MessagingGateway; import org.springframework.integration.annotation.Poller; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.DirectChannel; @@ -53,6 +60,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.support.ChannelInterceptor; @@ -66,6 +74,7 @@ import org.springframework.security.authentication.AuthenticationManager; import org.springframework.security.core.AuthenticationException; import org.springframework.security.core.context.SecurityContext; import org.springframework.security.core.context.SecurityContextHolder; +import org.springframework.security.task.DelegatingSecurityContextAsyncTaskExecutor; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -115,6 +124,8 @@ public class ChannelSecurityInterceptorSecuredChannelAnnotationTests { @Autowired TestHandler testConsumer; + @Autowired + TestGateway testGateway; @After public void tearDown() { @@ -252,6 +263,21 @@ public class ChannelSecurityInterceptorSecuredChannelAnnotationTests { instanceOf(AuthenticationCredentialsNotFoundException.class)); } + @Test + public void testSecurityContextPropagationAsyncGateway() throws Exception { + login("bob", "bobspassword", "ROLE_ADMIN", "ROLE_PRESIDENT"); + Future future = this.testGateway.test("foo"); + Message receive = this.securedChannelQueue.receive(10000); + assertNotNull(receive); + + MessageChannel replyChannel = receive.getHeaders().get(MessageHeaders.REPLY_CHANNEL, MessageChannel.class); + replyChannel.send(new GenericMessage<>("bar")); + + String result = future.get(10, TimeUnit.SECONDS); + assertNotNull(result); + assertEquals("bar", result); + } + private void login(String username, String password, String... roles) { SecurityContext context = SecurityTestUtils.createContext(username, password, roles); SecurityContextHolder.setContext(context); @@ -260,6 +286,7 @@ public class ChannelSecurityInterceptorSecuredChannelAnnotationTests { @Configuration @EnableIntegration + @IntegrationComponentScan @ImportResource("classpath:org/springframework/integration/security/config/commonSecurityConfiguration.xml") public static class ContextConfiguration { @@ -370,6 +397,19 @@ public class ChannelSecurityInterceptorSecuredChannelAnnotationTests { return channelSecurityInterceptor; } + @Bean + public AsyncTaskExecutor securityContextExecutor() { + return new DelegatingSecurityContextAsyncTaskExecutor(new SimpleAsyncTaskExecutor()); + } + + } + + @MessagingGateway(asyncExecutor = "securityContextExecutor") + public interface TestGateway { + + @Gateway(requestChannel = "queueChannel") + Future test(String payload); + } } diff --git a/src/reference/asciidoc/control-bus.adoc b/src/reference/asciidoc/control-bus.adoc index aa01898b95..85b4877043 100644 --- a/src/reference/asciidoc/control-bus.adoc +++ b/src/reference/asciidoc/control-bus.adoc @@ -15,13 +15,12 @@ you can specify an output channel if the result of the operation has a return va The Control Bus executes messages on the input channel as Spring Expression Language expressions. It takes a message, compiles the body to an expression, adds some context, and then executes it. -The default context supports any method that has been annotated with @ManagedAttribute or @ManagedOperation. -It also supports the methods on Spring's Lifecycle interface, and it supports methods that are used to configure several of Spring's TaskExecutor and TaskScheduler implementations. -The simplest way to ensure that your own methods are available to the Control Bus is to use the @ManagedAttribute and/or @ManagedOperation annotations. -Since those are also used for exposing methods to a JMX MBean registry, it's a convenient by-product (often the same -types of operations you want to expose to the Control Bus would be reasonable for exposing via JMX). +The default context supports any method that has been annotated with `@ManagedAttribute` or `@ManagedOperation`. +It also supports the methods on Spring's `Lifecycle` interface, and it supports methods that are used to configure several of Spring's `TaskExecutor` and `TaskScheduler` implementations. +The simplest way to ensure that your own methods are available to the Control Bus is to use the `@ManagedAttribute` and/or `@ManagedOperation` annotations. +Since those are also used for exposing methods to a JMX MBean registry, it's a convenient by-product (often the same types of operations you want to expose to the Control Bus would be reasonable for exposing via JMX). Resolution of any particular instance within the application context is achieved in the typical SpEL syntax. -Simply provide the bean name with the SpEL prefix for beans (@). +Simply provide the bean name with the SpEL prefix for beans (`@`). For example, to execute a method on a Spring Bean a client could send a message to the operation channel as follows: [source,java] @@ -30,5 +29,40 @@ Message operation = MessageBuilder.withPayload("@myServiceBean.shutdown()").buil operationChannel.send(operation) ---- -The root of the context for the expression is the `Message` itself, so you also have access to the 'payload' and 'headers' as variables within your expression. +The root of the context for the expression is the `Message` itself, so you also have access to the `payload` and `headers` as variables within your expression. This is consistent with all the other expression support in Spring Integration endpoints. + +With Java and Annotations the Control Bus can be configured as follows: + +[source,java] +---- +@Bean +@ServiceActivator(inputChannel = "operationChannel") +public ExpressionControlBusFactoryBean controlBus() { + return new ExpressionControlBusFactoryBean(); +} +---- + +Or, when using Java DSL flow definitions: + +[source,java] +---- +@Bean +public IntegrationFlow controlBusFlow() { + return IntegrationFlows.from("controlBus") + .controlBus() + .get(); +} +---- + +Or, if you prefer Lambda style with automatic `DirectChannel` creation: + +[source,java] +---- +@Bean +public IntegrationFlow controlBus() { + return IntegrationFlowDefinition::controlBus; +} +---- + +In this case, the channel is named `controlBus.input`. diff --git a/src/reference/asciidoc/security.adoc b/src/reference/asciidoc/security.adoc index 100f2e8f1b..4f54d3b014 100644 --- a/src/reference/asciidoc/security.adoc +++ b/src/reference/asciidoc/security.adoc @@ -88,7 +88,8 @@ public class ContextConfiguration { } @Bean - public ChannelSecurityInterceptor channelSecurityInterceptor(AuthenticationManager authenticationManager, + public ChannelSecurityInterceptor channelSecurityInterceptor( + AuthenticationManager authenticationManager, AccessDecisionManager accessDecisionManager) { ChannelSecurityInterceptor channelSecurityInterceptor = new ChannelSecurityInterceptor(); channelSecurityInterceptor.setAuthenticationManager(authenticationManager); @@ -111,7 +112,7 @@ 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. +sufficient 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). @@ -155,3 +156,31 @@ implementation to do the clean up operation to free the Thread in the end of inv 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. + +NOTE: When working with <>, you should use an appropriate `AbstractDelegatingSecurityContextSupport` implementation from Spring Security http://docs.spring.io/spring-security/site/docs/current/reference/html/concurrency.html[Concurrency Support], when security context propagation should be ensured over gateway invocation: + +[source,java] +---- +@Configuration +@EnableIntegration +@IntegrationComponentScan +public class ContextConfiguration { + + @Bean + public AsyncTaskExecutor securityContextExecutor() { + return new DelegatingSecurityContextAsyncTaskExecutor( + new SimpleAsyncTaskExecutor()); + } + +} + +... + +@MessagingGateway(asyncExecutor = "securityContextExecutor") +public interface SecuredGateway { + + @Gateway(requestChannel = "queueChannel") + Future send(String payload); + +} +---- \ No newline at end of file diff --git a/src/reference/asciidoc/stomp.adoc b/src/reference/asciidoc/stomp.adoc index 003a74523f..1a43362f82 100644 --- a/src/reference/asciidoc/stomp.adoc +++ b/src/reference/asciidoc/stomp.adoc @@ -20,7 +20,7 @@ The Spring Framework provides these implementations: * `WebSocketStompClient` - built on the Spring WebSocket API with support for standard JSR-356 WebSocket, Jetty 9, as well as SockJS for HTTP-based WebSocket emulation with SockJS Client. -* `Reactor2TcpStompClient` - built on `NettyTcpClient` from the `reactor-net` project. +* `ReactorNettyTcpStompClient` - built on `ReactorNettyTcpClient` from the `reactor-netty` project. Any other `StompClientSupport` implementation can be provided. See the JavaDocs of those classes for more information. @@ -121,8 +121,8 @@ A comprehensive Java & Annotation Configuration for STOMP Adapters may look like public class StompConfiguration { @Bean - public Reactor2TcpStompClient stompClient() { - Reactor2TcpStompClient stompClient = new Reactor2TcpStompClient("127.0.0.1", 61613); + public ReactorNettyTcpStompClient stompClient() { + ReactorNettyTcpStompClient stompClient = new ReactorNettyTcpStompClient("127.0.0.1", 61613); stompClient.setMessageConverter(new PassThruMessageConverter()); ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); taskScheduler.afterPropertiesSet(); @@ -133,7 +133,7 @@ public class StompConfiguration { @Bean public StompSessionManager stompSessionManager() { - Reactor2TcpStompSessionManager stompSessionManager = new Reactor2TcpStompSessionManager(stompClient()); + ReactorNettyTcpStompSessionManager stompSessionManager = new ReactorNettyTcpStompSessionManager(stompClient()); stompSessionManager.setAutoReceipt(true); return stompSessionManager; } diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 4ef4b8a167..e60b0f1525 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -17,6 +17,9 @@ See <> for more information. [[x5.0-general]] === General Changes +Spring Integration is now fully based on Spring Framework `5.0` and Project Reactor `3.0`. +Previous Project Reactor versions are no longer supported. + ==== Core Changes The `@Poller` annotation now has the `errorChannel` attribute for easier configuration of the underlying `MessagePublishingErrorHandler`. @@ -107,3 +110,10 @@ Inbound messages are now mapped with headers `RECEIVED_TOPIC`, `RECEIVED_QOS` an The outbound channel adapter now supports expressions for the topic, qos and retained properties; the defaults remain the same. See <> for more information. + +==== STOMP Changes + +The STOMP module has been changed to use `ReactorNettyTcpStompClient`, based on the Project Reactor `3.0` and `reactor-netty` extension. +The `Reactor2TcpStompSessionManager` has been renamed to the `ReactorNettyTcpStompSessionManager` according to the `ReactorNettyTcpStompClient` foundation. + +See <> for more information.