From 4d8494ae762b684e47f6f8abdc3c7239950fcfeb Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 7 May 2018 12:08:44 -0400 Subject: [PATCH] INT-4462: WebFluxInbound: cope with empty body JIRA: https://jira.spring.io/browse/INT-4462 When the HTTP request body is empty, the `HttpMessageReader` ends up with the empty `Mono` which can't be evaluated to any reasonable value. * Add fallback to `requestParams` when `Mono` for body is empty and also when `payloadExpression` returns null **Cherry-pick to 5.0.x** --- .../inbound/WebFluxInboundEndpoint.java | 9 +++--- .../inbound/WebFluxInboundEndpointTests.java | 29 ++++++++++++++++++- src/reference/asciidoc/webflux.adoc | 4 ++- 3 files changed, 36 insertions(+), 6 deletions(-) diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java index 010d7efd13..9a4ce4bdaa 100644 --- a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java @@ -150,6 +150,7 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W private Mono doHandle(ServerWebExchange exchange) { return extractRequestBody(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(requestMessage -> { @@ -267,10 +268,6 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W Object payload; if (getPayloadExpression() != null) { payload = getPayloadExpression().getValue(evaluationContext); - if (payload == null) { - throw new IllegalStateException("The payload expression '" + getPayloadExpression().getExpressionString() - + "' returned null."); - } } else { payload = httpEntity.getBody(); @@ -288,6 +285,10 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W } } + if (payload == null) { + payload = requestParams; + } + AbstractIntegrationMessageBuilder messageBuilder; if (payload instanceof Message) { diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpointTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpointTests.java index 124199b6ef..a135c00b6d 100644 --- a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpointTests.java +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpointTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 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. @@ -25,6 +25,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.http.HttpMethod; import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; import org.springframework.http.ResponseEntity; @@ -95,6 +96,16 @@ public class WebFluxInboundEndpointTests { .is5xxServerError(); } + @Test + public void testPostWithEmptyBody() { + this.webTestClient + .post() + .uri("/post?foo=foo") + .exchange() + .expectStatus().isOk() + .expectBody(String.class).isEqualTo("{foo=[foo]}"); + } + @Configuration @EnableWebFlux @EnableIntegration @@ -156,6 +167,22 @@ public class WebFluxInboundEndpointTests { return new ResponseEntity<>("<500 Internal Server Error,{}>", HttpStatus.INTERNAL_SERVER_ERROR); } + @Bean + public WebFluxInboundEndpoint postInboundEndpoint() { + WebFluxInboundEndpoint endpoint = new WebFluxInboundEndpoint(); + RequestMapping requestMapping = new RequestMapping(); + requestMapping.setPathPatterns("/post"); + requestMapping.setMethods(HttpMethod.POST); + endpoint.setRequestMapping(requestMapping); + endpoint.setRequestChannelName("postServiceChannel"); + return endpoint; + } + + @ServiceActivator(inputChannel = "postServiceChannel") + String service(Object payload) { + return payload.toString(); + } + } diff --git a/src/reference/asciidoc/webflux.adoc b/src/reference/asciidoc/webflux.adoc index 4abf0d3d63..d0674dd583 100644 --- a/src/reference/asciidoc/webflux.adoc +++ b/src/reference/asciidoc/webflux.adoc @@ -6,7 +6,7 @@ The WebFlux Spring Integration module (`spring-integration-webflux`) allows for the execution of HTTP requests and the processing of inbound HTTP requests in Reactive manner. The WebFlux support consists of the following gateway implementations: `WebFluxInboundEndpoint`, `WebFluxRequestExecutingMessageHandler`. -The implementation is fully based on the Spring http://docs.spring.io/spring/docs/5.0.0.RC3/spring-framework-reference/web.html#web-reactive[WebFlux] and https://projectreactor.io/[Project Reactor] foundations. +The implementation is fully based on the Spring https://docs.spring.io/spring/docs/current/spring-framework-reference/web-reactive.html#spring-webflux[WebFlux] and https://projectreactor.io/[Project Reactor] foundations. Also see <> for more information since many options are shared between reactive and regular HTTP components. [[webflux-inbound]] @@ -65,6 +65,8 @@ public IntegrationFlow sseFlow() { Also see <> and <> for more possible configuration options. +When the request body is empty, or `payloadExpression` returns `null`, the request params `MultiValueMap` is used for a `payload` of the target message to process. + [[webflux-outbound]] === WebFlux Outbound Components