From 7ee19ffe800aef9b368ec123702385b9f452c845 Mon Sep 17 00:00:00 2001 From: spencergibb Date: Mon, 18 May 2020 17:08:12 -0400 Subject: [PATCH] Wraps ServerHttpRequests even if there is no body. This avoids multiple subscriptions. Fixes gh-1675 --- .../support/ServerWebExchangeUtils.java | 70 +++++++++++-------- ...yGatewayFilterFactoryIntegrationTests.java | 13 ++++ 2 files changed, 54 insertions(+), 29 deletions(-) diff --git a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/ServerWebExchangeUtils.java b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/ServerWebExchangeUtils.java index 52f420ab..62cbfc86 100644 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/ServerWebExchangeUtils.java +++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/ServerWebExchangeUtils.java @@ -23,6 +23,7 @@ import java.util.Map; import java.util.function.Function; import java.util.function.Predicate; +import io.netty.buffer.EmptyByteBuf; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import reactor.core.publisher.Flux; @@ -34,10 +35,12 @@ import org.springframework.cloud.gateway.handler.predicate.RoutePredicateFactory import org.springframework.core.io.buffer.DataBuffer; import org.springframework.core.io.buffer.DataBufferUtils; import org.springframework.core.io.buffer.NettyDataBuffer; +import org.springframework.core.io.buffer.NettyDataBufferFactory; import org.springframework.http.HttpStatus; import org.springframework.http.server.reactive.AbstractServerHttpResponse; import org.springframework.http.server.reactive.ServerHttpRequest; import org.springframework.http.server.reactive.ServerHttpRequestDecorator; +import org.springframework.http.server.reactive.ServerHttpResponse; import org.springframework.util.Assert; import org.springframework.web.server.ServerWebExchange; import org.springframework.web.util.UriComponentsBuilder; @@ -343,38 +346,47 @@ public final class ServerWebExchangeUtils { private static Mono cacheRequestBody(ServerWebExchange exchange, boolean cacheDecoratedRequest, Function> function) { + ServerHttpResponse response = exchange.getResponse(); + NettyDataBufferFactory factory = (NettyDataBufferFactory) response + .bufferFactory(); // Join all the DataBuffers so we have a single DataBuffer for the body - return DataBufferUtils.join(exchange.getRequest().getBody()).map(dataBuffer -> { - if (dataBuffer.readableByteCount() > 0) { - if (log.isTraceEnabled()) { - log.trace("retaining body in exchange attribute"); - } - exchange.getAttributes().put(CACHED_REQUEST_BODY_ATTR, dataBuffer); - } + return DataBufferUtils.join(exchange.getRequest().getBody()) + .defaultIfEmpty( + factory.wrap(new EmptyByteBuf(factory.getByteBufAllocator()))) + .map(dataBuffer -> decorate(exchange, dataBuffer, cacheDecoratedRequest)) + .switchIfEmpty(Mono.just(exchange.getRequest())).flatMap(function); + } - ServerHttpRequest decorator = new ServerHttpRequestDecorator( - exchange.getRequest()) { - @Override - public Flux getBody() { - return Mono.fromSupplier(() -> { - if (exchange.getAttributeOrDefault(CACHED_REQUEST_BODY_ATTR, - null) == null) { - // probably == downstream closed - return null; - } - // TODO: deal with Netty - NettyDataBuffer pdb = (NettyDataBuffer) dataBuffer; - return pdb.factory().wrap(pdb.getNativeBuffer().retainedSlice()); - }).flux(); - } - }; - if (cacheDecoratedRequest) { - exchange.getAttributes().put(CACHED_SERVER_HTTP_REQUEST_DECORATOR_ATTR, - decorator); + private static ServerHttpRequest decorate(ServerWebExchange exchange, + DataBuffer dataBuffer, boolean cacheDecoratedRequest) { + if (dataBuffer.readableByteCount() > 0) { + if (log.isTraceEnabled()) { + log.trace("retaining body in exchange attribute"); } - return decorator; - // return function.apply(decorator)/*.then(monoVoid())*/; - }).switchIfEmpty(Mono.just(exchange.getRequest())).flatMap(function); + exchange.getAttributes().put(CACHED_REQUEST_BODY_ATTR, dataBuffer); + } + + ServerHttpRequest decorator = new ServerHttpRequestDecorator( + exchange.getRequest()) { + @Override + public Flux getBody() { + return Mono.fromSupplier(() -> { + if (exchange.getAttributeOrDefault(CACHED_REQUEST_BODY_ATTR, + null) == null) { + // probably == downstream closed or no body + return null; + } + // TODO: deal with Netty + NettyDataBuffer pdb = (NettyDataBuffer) dataBuffer; + return pdb.factory().wrap(pdb.getNativeBuffer().retainedSlice()); + }).flux(); + } + }; + if (cacheDecoratedRequest) { + exchange.getAttributes().put(CACHED_SERVER_HTTP_REQUEST_DECORATOR_ATTR, + decorator); + } + return decorator; } } diff --git a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/factory/RetryGatewayFilterFactoryIntegrationTests.java b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/factory/RetryGatewayFilterFactoryIntegrationTests.java index 1ca17cf3..4363db77 100644 --- a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/factory/RetryGatewayFilterFactoryIntegrationTests.java +++ b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/factory/RetryGatewayFilterFactoryIntegrationTests.java @@ -148,6 +148,7 @@ public class RetryGatewayFilterFactoryIntegrationTests extends BaseWebClientTest } @Test + // @Ignore // FIXME: https://github.com/spring-cloud/spring-cloud-gateway/issues/1675 public void shouldNotRetryWhenSleepyRequestPost() throws Exception { testClient.mutate().responseTimeout(Duration.ofSeconds(10)).build().post() .uri("/sleep?key=notRetriesSleepyRequestPost&millis=3000") @@ -158,6 +159,18 @@ public class RetryGatewayFilterFactoryIntegrationTests extends BaseWebClientTest .hasValue(1); } + @Test + public void shouldNotRetryWhenSleepyRequestPostWithBody() throws Exception { + testClient.mutate().responseTimeout(Duration.ofSeconds(10)).build().post() + .uri("/sleep?key=notRetriesSleepyRequestPostWithBody&millis=3000") + .header(HttpHeaders.HOST, "www.retry-only-get.org") + .bodyValue("retry sleepy post with body").exchange().expectStatus() + .isEqualTo(HttpStatus.GATEWAY_TIMEOUT); + + assertThat(TestConfig.map.get("notRetriesSleepyRequestPostWithBody")).isNotNull() + .hasValue(1); + } + @Test public void shouldRetryWhenSleepyRequestGet() throws Exception { testClient.mutate().responseTimeout(Duration.ofSeconds(10)).build().get()