diff --git a/.gitignore b/.gitignore index 62227c8c..dc791ba1 100644 --- a/.gitignore +++ b/.gitignore @@ -21,3 +21,4 @@ _site/ *.swo .vscode/ .flattened-pom.xml +.sdkmanrc 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 68ce8223..e9530b36 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; @@ -337,38 +340,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 c5dbed16..914b90b3 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 @@ -25,7 +25,6 @@ import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.hamcrest.CoreMatchers; -import org.junit.Ignore; import org.junit.Rule; import org.junit.Test; import org.junit.runner.RunWith; @@ -149,7 +148,7 @@ public class RetryGatewayFilterFactoryIntegrationTests extends BaseWebClientTest } @Test - @Ignore // FIXME: https://github.com/spring-cloud/spring-cloud-gateway/issues/1675 + // @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") @@ -160,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()