GH-2464: WebFlux: Get rid of Mono.block() (#2465)

* GH-2464: WebFlux: Get rid of Mono.block()

Fixes #2464

The `WebFluxInboundEndpoint` resolves a `Principal` via `Mono.block()`
operation.
This is prohibited situation in the non-blocking thread, like Reactor
Netty

* Defer `Mono<Principal>` resolution to the message header via
deferring the whole `Message` creating via `flatMap()` operation on the
main `doHandle()` `Mono`

**Cherry-pick to 5.0.x**

* * Fix Checkstyle for the Reactive Spring Security testing utils static imports
* Add `defaultIfEmpty()`, when `exchange.getPrincipal()` is an empty `Mono`

Author: artembilan

spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java

@@ -152,7 +152,7 @@ private Mono<Void> doHandle(ServerWebExchange exchange) {
 				.doOnSubscribe(s -> this.activeCount.incrementAndGet())
 				.switchIfEmpty(Mono.just(exchange.getRequest().getQueryParams()))
 				.map(body -> new HttpEntity<>(body, exchange.getRequest().getHeaders()))
-				.map(entity -> buildMessage(entity, exchange))
+				.flatMap(entity -> buildMessage(entity, exchange))
 				.flatMap(requestMessage -> {
 					if (this.expectReply) {
 						return sendAndReceiveMessageReactive(requestMessage)
@@ -235,7 +235,7 @@ else if (MediaType.MULTIPART_FORM_DATA.isCompatibleWith(contentType)) {
 	}
 
 	@SuppressWarnings("unchecked")
-	private Message<?> buildMessage(HttpEntity<?> httpEntity, ServerWebExchange exchange) {
+	private Mono<Message<?>> buildMessage(HttpEntity<?> httpEntity, ServerWebExchange exchange) {
 		ServerHttpRequest request = exchange.getRequest();
 		HttpHeaders requestHeaders = request.getHeaders();
 		Map<String, Object> exchangeAttributes = exchange.getAttributes();
@@ -289,12 +289,12 @@ private Message<?> buildMessage(HttpEntity<?> httpEntity, ServerWebExchange exch
 			payload = requestParams;
 		}
 
-		AbstractIntegrationMessageBuilder<?> messageBuilder;
+		AbstractIntegrationMessageBuilder<Object> messageBuilder;
 
 		if (payload instanceof Message<?>) {
 			messageBuilder =
 					getMessageBuilderFactory()
-							.fromMessage((Message<?>) payload)
+							.fromMessage((Message<Object>) payload)
 							.copyHeadersIfAbsent(headers);
 		}
 		else {
@@ -304,11 +304,17 @@ private Message<?> buildMessage(HttpEntity<?> httpEntity, ServerWebExchange exch
 							.copyHeaders(headers);
 		}
 
-		return messageBuilder
+		messageBuilder
 				.setHeader(org.springframework.integration.http.HttpHeaders.REQUEST_URL, request.getURI().toString())
-				.setHeader(org.springframework.integration.http.HttpHeaders.REQUEST_METHOD, request.getMethod().toString())
-				.setHeader(org.springframework.integration.http.HttpHeaders.USER_PRINCIPAL, exchange.getPrincipal().block())
-				.build();
+				.setHeader(org.springframework.integration.http.HttpHeaders.REQUEST_METHOD,
+						request.getMethod().toString());
+
+		return exchange.getPrincipal()
+				.map(principal ->
+						messageBuilder
+								.setHeader(org.springframework.integration.http.HttpHeaders.USER_PRINCIPAL, principal))
+				.defaultIfEmpty(messageBuilder)
+				.map(AbstractIntegrationMessageBuilder::build);
 	}
 
 	private Mono<Void> populateResponse(ServerWebExchange exchange, Message<?> replyMessage) {

spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java

@@ -17,13 +17,16 @@
 package org.springframework.integration.webflux.dsl;
 
 import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.is;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThat;
 import static org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors.httpBasic;
-import static org.springframework.security.test.web.servlet.setup.SecurityMockMvcConfigurers.springSecurity;
 import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
+import static org.springframework.web.reactive.function.client.ExchangeFilterFunctions.Credentials.basicAuthenticationCredentials;
+import static org.springframework.web.reactive.function.client.ExchangeFilterFunctions.basicAuthentication;
 
+import java.security.Principal;
 import java.util.Collections;
 
 import javax.annotation.Resource;
@@ -48,6 +51,7 @@
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.http.HttpHeaders;
 import org.springframework.integration.http.dsl.Http;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.webflux.outbound.WebFluxRequestExecutingMessageHandler;
@@ -60,10 +64,18 @@
 import org.springframework.security.config.annotation.web.builders.HttpSecurity;
 import org.springframework.security.config.annotation.web.configuration.EnableWebSecurity;
 import org.springframework.security.config.annotation.web.configuration.WebSecurityConfigurerAdapter;
+import org.springframework.security.config.annotation.web.reactive.EnableWebFluxSecurity;
+import org.springframework.security.config.web.server.ServerHttpSecurity;
+import org.springframework.security.core.userdetails.MapReactiveUserDetailsService;
+import org.springframework.security.core.userdetails.ReactiveUserDetailsService;
 import org.springframework.security.core.userdetails.User;
+import org.springframework.security.core.userdetails.UserDetails;
 import org.springframework.security.core.userdetails.UserDetailsService;
 import org.springframework.security.crypto.factory.PasswordEncoderFactories;
 import org.springframework.security.provisioning.InMemoryUserDetailsManager;
+import org.springframework.security.test.web.reactive.server.SecurityMockServerConfigurers;
+import org.springframework.security.test.web.servlet.setup.SecurityMockMvcConfigurers;
+import org.springframework.security.web.server.SecurityWebFilterChain;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit4.SpringRunner;
 import org.springframework.test.context.web.WebAppConfiguration;
@@ -116,11 +128,14 @@ public class WebFluxDslTests {
 	public void setup() {
 		this.mockMvc =
 				MockMvcBuilders.webAppContextSetup(this.wac)
-						.apply(springSecurity())
+						.apply(SecurityMockMvcConfigurers.springSecurity())
 						.build();
 
 		this.webTestClient =
 				WebTestClient.bindToApplicationContext(this.wac)
+						.apply(SecurityMockServerConfigurers.springSecurity())
+						.configureClient()
+						.filter(basicAuthentication())
 						.build();
 	}
 
@@ -198,6 +213,7 @@ public void testHttpReactiveProxyFlow() throws Exception {
 	@SuppressWarnings("unchecked")
 	public void testHttpReactivePost() {
 		this.webTestClient.post().uri("/reactivePost")
+				.attributes(basicAuthenticationCredentials("guest", "guest"))
 				.body(Mono.just("foo\nbar\nbaz"), String.class)
 				.exchange()
 				.expectStatus().isAccepted();
@@ -206,6 +222,8 @@ public void testHttpReactivePost() {
 		assertNotNull(store);
 		assertThat(store.getPayload(), instanceOf(Flux.class));
 
+		assertThat(store.getHeaders().get(HttpHeaders.USER_PRINCIPAL, Principal.class).getName(), is("guest"));
+
 		StepVerifier
 				.create((Publisher<String>) store.getPayload())
 				.expectNext("foo", "bar", "baz")
@@ -217,6 +235,7 @@ public void testHttpReactivePost() {
 	public void testSse() {
 		Flux<String> responseBody =
 				this.webTestClient.get().uri("/sse")
+						.attributes(basicAuthenticationCredentials("guest", "guest"))
 						.exchange()
 						.returnResult(String.class)
 						.getResponseBody();
@@ -230,22 +249,40 @@ public void testSse() {
 	@Configuration
 	@EnableWebFlux
 	@EnableWebSecurity
+	@EnableWebFluxSecurity
 	@EnableIntegration
 	public static class ContextConfiguration extends WebSecurityConfigurerAdapter {
 
+		@Bean
+		public UserDetails userDetails() {
+			return User.withUsername("guest")
+					.passwordEncoder(PasswordEncoderFactories.createDelegatingPasswordEncoder()::encode)
+					.password("guest")
+					.roles("ADMIN")
+					.build();
+		}
+
 		@Override
 		@Bean
 		public UserDetailsService userDetailsService() {
-			InMemoryUserDetailsManager manager = new InMemoryUserDetailsManager();
+			return new InMemoryUserDetailsManager(userDetails());
+		}
 
-			manager.createUser(
-					User.withUsername("guest")
-							.passwordEncoder(PasswordEncoderFactories.createDelegatingPasswordEncoder()::encode)
-							.password("guest")
-							.roles("ADMIN")
-							.build());
+		@Bean
+		public ReactiveUserDetailsService reactiveUserDetailsService() {
+			return new MapReactiveUserDetailsService(userDetails());
+		}
 
-			return manager;
+
+		@Bean
+		public SecurityWebFilterChain reactiveSpringSecurityFilterChain(ServerHttpSecurity http) {
+			return http.authorizeExchange()
+					.anyExchange().hasRole("ADMIN")
+					.and()
+					.httpBasic()
+					.and()
+					.csrf().disable()
+					.build();
 		}
 
 		@Override

src/checkstyle/checkstyle.xml

@@ -104,7 +104,9 @@
 				org.springframework.test.web.servlet.request.MockMvcRequestBuilders.*,
 				org.springframework.test.web.servlet.result.MockMvcResultMatchers.*,
 				org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors.*,
-				org.springframework.security.test.web.servlet.setup.SecurityMockMvcConfigurers.*" />
+				org.springframework.security.test.web.servlet.setup.SecurityMockMvcConfigurers.*,
+				org.springframework.web.reactive.function.client.ExchangeFilterFunctions.*,
+				org.springframework.web.reactive.function.client.ExchangeFilterFunctions.Credentials.*"	/>
 		</module>
 		<module name="IllegalImport" />
 		<module name="RedundantImport" />