Improve Security and Control Bus Docs

Fixes https://github.com/spring-projects/spring-integration-samples/issues/189

Also see http://stackoverflow.com/questions/41403174/how-to-propagate-spring-security-context-in-spring-integration-async-messaging-g

* Add Control Bus Java DSL and Annotation configuration sample to the Docs
* Mention Spring Security  `DelegatingSecurityContextAsyncTaskExecutor` in Docs and add test-case to demonstrate Security Context propagation via `@MessagingGateway`

Reflect reality for STOMP Docs

Doc Polishing

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<String> 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<String> test(String payload);
+
 	}
 
 }

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`.

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 <<async-gateway,Asynchronous Gateway>>, 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<String> send(String payload);
+
+}
+----

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;
     }

src/reference/asciidoc/whats-new.adoc

@@ -17,6 +17,9 @@ See <<mongodb-outbound-gateway>> 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 <<mqtt>> 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 <<stomp>> for more information.