From c46543b3568b935de06208161182c4daf8ad5a02 Mon Sep 17 00:00:00 2001 From: Spencer Gibb Date: Thu, 13 Jun 2019 16:36:14 -0400 Subject: [PATCH 1/2] Only add headers if response is not committed. --- .../cloud/gateway/test/BaseWebClientTests.java | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/BaseWebClientTests.java b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/BaseWebClientTests.java index 30573961..ff944508 100644 --- a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/BaseWebClientTests.java +++ b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/BaseWebClientTests.java @@ -97,11 +97,15 @@ public class BaseWebClientTests { log.info("modifyResponseFilter start"); String value = exchange.getAttributeOrDefault(GATEWAY_HANDLER_MAPPER_ATTR, "N/A"); - exchange.getResponse().getHeaders().add(HANDLER_MAPPER_HEADER, value); + if (!exchange.getResponse().isCommitted()) { + exchange.getResponse().getHeaders().add(HANDLER_MAPPER_HEADER, value); + } Route route = exchange.getAttributeOrDefault(GATEWAY_ROUTE_ATTR, null); if (route != null) { - exchange.getResponse().getHeaders().add(ROUTE_ID_HEADER, - route.getId()); + if (!exchange.getResponse().isCommitted()) { + exchange.getResponse().getHeaders().add(ROUTE_ID_HEADER, + route.getId()); + } } return chain.filter(exchange); }; From ec0d906ff3992789a79ab6e3029374dae478aea8 Mon Sep 17 00:00:00 2001 From: Spencer Gibb Date: Wed, 19 Jun 2019 14:40:06 -0400 Subject: [PATCH 2/2] Adds ServerWebExchangeUtils methods for caching the request body. Adapts the methods from AlwaysRetainBodyGlobalFilter to a generic way to cache the request body. That filter was removed and AdaptCachedBodyGlobalFilter and ReadBodyPredicateFactory were updated to use the new methods. fixes gh-946 --- .../config/GatewayAutoConfiguration.java | 21 +--- .../filter/AdaptCachedBodyGlobalFilter.java | 62 ++++++++--- .../filter/AlwaysRetainBodyGlobalFilter.java | 101 ----------------- .../filter/RemoveCachedBodyFilter.java | 10 +- .../predicate/ReadBodyPredicateFactory.java | 56 +++------- .../support/ServerWebExchangeUtils.java | 103 ++++++++++++++++++ ...yGatewayFilterFactoryIntegrationTests.java | 4 +- .../ReadBodyPredicateFactoryTest.java | 42 +++---- 8 files changed, 194 insertions(+), 205 deletions(-) delete mode 100644 spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/AlwaysRetainBodyGlobalFilter.java diff --git a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/config/GatewayAutoConfiguration.java b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/config/GatewayAutoConfiguration.java index b1221eca..e848a5fd 100644 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/config/GatewayAutoConfiguration.java +++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/config/GatewayAutoConfiguration.java @@ -45,7 +45,6 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties import org.springframework.boot.context.properties.PropertyMapper; import org.springframework.cloud.gateway.actuate.GatewayControllerEndpoint; import org.springframework.cloud.gateway.filter.AdaptCachedBodyGlobalFilter; -import org.springframework.cloud.gateway.filter.AlwaysRetainBodyGlobalFilter; import org.springframework.cloud.gateway.filter.ForwardPathFilter; import org.springframework.cloud.gateway.filter.ForwardRoutingFilter; import org.springframework.cloud.gateway.filter.GlobalFilter; @@ -126,7 +125,6 @@ import org.springframework.context.annotation.DependsOn; import org.springframework.context.annotation.Primary; import org.springframework.core.convert.ConversionService; import org.springframework.core.env.Environment; -import org.springframework.http.codec.ServerCodecConfigurer; import org.springframework.util.StringUtils; import org.springframework.validation.Validator; import org.springframework.web.reactive.DispatcherHandler; @@ -263,11 +261,6 @@ public class GatewayAutoConfiguration { return new AdaptCachedBodyGlobalFilter(); } - @Bean - public AlwaysRetainBodyGlobalFilter alwaysRetainBodyGlobalFilter() { - return new AlwaysRetainBodyGlobalFilter(); - } - @Bean public RemoveCachedBodyFilter removeCachedBodyFilter() { return new RemoveCachedBodyFilter(); @@ -403,9 +396,8 @@ public class GatewayAutoConfiguration { } @Bean - public ModifyRequestBodyGatewayFilterFactory modifyRequestBodyGatewayFilterFactory( - ServerCodecConfigurer codecConfigurer) { - return new ModifyRequestBodyGatewayFilterFactory(codecConfigurer); + public ModifyRequestBodyGatewayFilterFactory modifyRequestBodyGatewayFilterFactory() { + return new ModifyRequestBodyGatewayFilterFactory(); } @Bean @@ -414,9 +406,8 @@ public class GatewayAutoConfiguration { } @Bean - public ModifyResponseBodyGatewayFilterFactory modifyResponseBodyGatewayFilterFactory( - ServerCodecConfigurer codecConfigurer) { - return new ModifyResponseBodyGatewayFilterFactory(codecConfigurer); + public ModifyResponseBodyGatewayFilterFactory modifyResponseBodyGatewayFilterFactory() { + return new ModifyResponseBodyGatewayFilterFactory(); } @Bean @@ -600,8 +591,8 @@ public class GatewayAutoConfiguration { }); } - //TODO: add configuration to turn on wiretap - //httpClient = httpClient.wiretap(true); + // TODO: add configuration to turn on wiretap + // httpClient = httpClient.wiretap(true); return httpClient; } diff --git a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/AdaptCachedBodyGlobalFilter.java b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/AdaptCachedBodyGlobalFilter.java index c463bf31..f505f406 100644 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/AdaptCachedBodyGlobalFilter.java +++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/AdaptCachedBodyGlobalFilter.java @@ -16,39 +16,65 @@ package org.springframework.cloud.gateway.filter; -import reactor.core.publisher.Flux; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + import reactor.core.publisher.Mono; +import org.springframework.cloud.gateway.event.EnableBodyCachingEvent; +import org.springframework.cloud.gateway.route.Route; +import org.springframework.cloud.gateway.support.ServerWebExchangeUtils; +import org.springframework.context.ApplicationListener; import org.springframework.core.Ordered; import org.springframework.core.io.buffer.DataBuffer; -import org.springframework.http.server.reactive.ServerHttpRequestDecorator; +import org.springframework.http.server.reactive.ServerHttpRequest; import org.springframework.web.server.ServerWebExchange; -public class AdaptCachedBodyGlobalFilter implements GlobalFilter, Ordered { +import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.CACHED_REQUEST_BODY_ATTR; +import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.CACHED_SERVER_HTTP_REQUEST_DECORATOR_ATTR; +import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.GATEWAY_ROUTE_ATTR; + +public class AdaptCachedBodyGlobalFilter + implements GlobalFilter, Ordered, ApplicationListener { + + private ConcurrentMap routesToCache = new ConcurrentHashMap<>(); /** * Cached request body key. */ - public static final String CACHED_REQUEST_BODY_KEY = "cachedRequestBody"; + @Deprecated + public static final String CACHED_REQUEST_BODY_KEY = CACHED_REQUEST_BODY_ATTR; + + @Override + public void onApplicationEvent(EnableBodyCachingEvent event) { + this.routesToCache.putIfAbsent(event.getRouteId(), true); + } @Override public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) { - - Flux body = exchange.getAttributeOrDefault(CACHED_REQUEST_BODY_KEY, - null); - if (body != null) { - ServerHttpRequestDecorator decorator = new ServerHttpRequestDecorator( - exchange.getRequest()) { - @Override - public Flux getBody() { - return body; - } - }; - exchange.getAttributes().remove(CACHED_REQUEST_BODY_KEY); - return chain.filter(exchange.mutate().request(decorator).build()); + // the cached ServerHttpRequest is used when the ServerWebExchange can not be + // mutated, for example, during a predicate where the body is read, but still + // needs to be cached. + ServerHttpRequest cachedRequest = exchange + .getAttributeOrDefault(CACHED_SERVER_HTTP_REQUEST_DECORATOR_ATTR, null); + if (cachedRequest != null) { + exchange.getAttributes().remove(CACHED_SERVER_HTTP_REQUEST_DECORATOR_ATTR); + return chain.filter(exchange.mutate().request(cachedRequest).build()); } - return chain.filter(exchange); + // + DataBuffer body = exchange.getAttributeOrDefault(CACHED_REQUEST_BODY_ATTR, null); + Route route = exchange.getAttribute(GATEWAY_ROUTE_ATTR); + + if (body != null || !this.routesToCache.containsKey(route.getId())) { + return chain.filter(exchange); + } + + return ServerWebExchangeUtils + .cacheRequestBody(exchange, + (serverHttpRequest) -> chain.filter( + exchange.mutate().request(serverHttpRequest).build())) + .switchIfEmpty(chain.filter(exchange)); } @Override diff --git a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/AlwaysRetainBodyGlobalFilter.java b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/AlwaysRetainBodyGlobalFilter.java deleted file mode 100644 index d5d4f78c..00000000 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/AlwaysRetainBodyGlobalFilter.java +++ /dev/null @@ -1,101 +0,0 @@ -/* - * Copyright 2013-2019 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. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.gateway.filter; - -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; - -import org.springframework.cloud.gateway.event.EnableBodyCachingEvent; -import org.springframework.cloud.gateway.route.Route; -import org.springframework.context.ApplicationListener; -import org.springframework.core.Ordered; -import org.springframework.core.io.buffer.DataBuffer; -import org.springframework.core.io.buffer.DataBufferUtils; -import org.springframework.core.io.buffer.NettyDataBuffer; -import org.springframework.http.server.reactive.ServerHttpRequestDecorator; -import org.springframework.web.server.ServerWebExchange; - -import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.GATEWAY_ROUTE_ATTR; - -public class AlwaysRetainBodyGlobalFilter - implements GlobalFilter, Ordered, ApplicationListener { - - private static final Log log = LogFactory.getLog(AlwaysRetainBodyGlobalFilter.class); - - private ConcurrentMap routesToCache = new ConcurrentHashMap<>(); - - /** - * Request body cache key. - */ - public static final String ALWAYS_CACHE_REQUEST_BODY_KEY = "alwaysCacheRequestBody"; - - @Override - public void onApplicationEvent(EnableBodyCachingEvent event) { - this.routesToCache.putIfAbsent(event.getRouteId(), true); - } - - @Override - public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) { - Object body = exchange.getAttributeOrDefault(ALWAYS_CACHE_REQUEST_BODY_KEY, null); - Route route = exchange.getAttribute(GATEWAY_ROUTE_ATTR); - - if (body != null || !this.routesToCache.containsKey(route.getId())) { - return chain.filter(exchange); - } - - return DataBufferUtils.join(exchange.getRequest().getBody()) - .flatMap(dataBuffer -> { - if (dataBuffer.readableByteCount() > 0) { - if (log.isTraceEnabled()) { - log.trace("retaining body in exchange attribute"); - } - exchange.getAttributes().put(ALWAYS_CACHE_REQUEST_BODY_KEY, - dataBuffer); - } - - ServerHttpRequestDecorator decorator = new ServerHttpRequestDecorator( - exchange.getRequest()) { - @Override - public Flux getBody() { - return Mono.fromSupplier(() -> { - if (exchange.getAttributeOrDefault( - ALWAYS_CACHE_REQUEST_BODY_KEY, null) == null) { - // probably == downstream closed - return null; - } - // TODO: deal with Netty - NettyDataBuffer pdb = (NettyDataBuffer) dataBuffer; - return pdb.factory() - .wrap(pdb.getNativeBuffer().retainedSlice()); - }).flux(); - } - }; - return chain.filter(exchange.mutate().request(decorator).build()); - }).switchIfEmpty(chain.filter(exchange)); - } - - @Override - public int getOrder() { - return -10; - } - -} diff --git a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/RemoveCachedBodyFilter.java b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/RemoveCachedBodyFilter.java index 9411cb3d..25a390c2 100644 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/RemoveCachedBodyFilter.java +++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/RemoveCachedBodyFilter.java @@ -24,7 +24,7 @@ import org.springframework.core.Ordered; import org.springframework.core.io.buffer.PooledDataBuffer; import org.springframework.web.server.ServerWebExchange; -import static org.springframework.cloud.gateway.filter.AlwaysRetainBodyGlobalFilter.ALWAYS_CACHE_REQUEST_BODY_KEY; +import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.CACHED_REQUEST_BODY_ATTR; public class RemoveCachedBodyFilter implements GlobalFilter, Ordered { @@ -33,13 +33,13 @@ public class RemoveCachedBodyFilter implements GlobalFilter, Ordered { @Override public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) { return chain.filter(exchange).doFinally(s -> { - PooledDataBuffer b = (PooledDataBuffer) exchange.getAttributes() - .remove(ALWAYS_CACHE_REQUEST_BODY_KEY); - if (b != null && b.isAllocated()) { + PooledDataBuffer dataBuffer = (PooledDataBuffer) exchange.getAttributes() + .remove(CACHED_REQUEST_BODY_ATTR); + if (dataBuffer != null && dataBuffer.isAllocated()) { if (log.isTraceEnabled()) { log.trace("releasing cached body in exchange attribute"); } - b.release(); + dataBuffer.release(); } }); } diff --git a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/handler/predicate/ReadBodyPredicateFactory.java b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/handler/predicate/ReadBodyPredicateFactory.java index 12384b77..ece0aeb9 100644 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/handler/predicate/ReadBodyPredicateFactory.java +++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/handler/predicate/ReadBodyPredicateFactory.java @@ -22,21 +22,15 @@ import java.util.function.Predicate; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import org.springframework.cloud.gateway.handler.AsyncPredicate; -import org.springframework.core.io.buffer.DataBuffer; -import org.springframework.core.io.buffer.DataBufferUtils; +import org.springframework.cloud.gateway.support.ServerWebExchangeUtils; import org.springframework.http.codec.HttpMessageReader; -import org.springframework.http.server.reactive.ServerHttpRequest; -import org.springframework.http.server.reactive.ServerHttpRequestDecorator; import org.springframework.web.reactive.function.server.HandlerStrategies; import org.springframework.web.reactive.function.server.ServerRequest; import org.springframework.web.server.ServerWebExchange; -import static org.springframework.cloud.gateway.filter.AdaptCachedBodyGlobalFilter.CACHED_REQUEST_BODY_KEY; - /** * Predicate that reads the body and applies a user provided predicate to run on the body. * The body is cached in memory so that possible subsequent calls to the predicate do not @@ -45,7 +39,7 @@ import static org.springframework.cloud.gateway.filter.AdaptCachedBodyGlobalFilt public class ReadBodyPredicateFactory extends AbstractRoutePredicateFactory { - protected static final Log LOGGER = LogFactory.getLog(ReadBodyPredicateFactory.class); + protected static final Log log = LogFactory.getLog(ReadBodyPredicateFactory.class); private static final String TEST_ATTRIBUTE = "read_body_predicate_test_attribute"; @@ -78,47 +72,23 @@ public class ReadBodyPredicateFactory return Mono.just(test); } catch (ClassCastException e) { - if (LOGGER.isDebugEnabled()) { - LOGGER.debug("Predicate test failed because class in predicate " + if (log.isDebugEnabled()) { + log.debug("Predicate test failed because class in predicate " + "does not match the cached body object", e); } } return Mono.just(false); } else { - // Join all the DataBuffers so we have a single DataBuffer for the body - return DataBufferUtils.join(exchange.getRequest().getBody()) - .flatMap(dataBuffer -> { - byte[] bytes = new byte[dataBuffer.readableByteCount()]; - dataBuffer.read(bytes); - DataBufferUtils.release(dataBuffer); - Flux cachedFlux = Flux.defer(() -> { - DataBuffer buffer = exchange.getResponse().bufferFactory() - .wrap(bytes); - DataBufferUtils.retain(buffer); - return Mono.just(buffer); - }); - - ServerHttpRequest mutatedRequest = new ServerHttpRequestDecorator( - exchange.getRequest()) { - @Override - public Flux getBody() { - return cachedFlux; - } - }; - return ServerRequest - .create(exchange.mutate().request(mutatedRequest) - .build(), messageReaders) - .bodyToMono(inClass).doOnNext(objectValue -> { - exchange.getAttributes().put( - CACHE_REQUEST_BODY_OBJECT_KEY, - objectValue); - exchange.getAttributes() - .put(CACHED_REQUEST_BODY_KEY, cachedFlux); - }).map(objectValue -> config.predicate - .test(objectValue)); - }); - + return ServerWebExchangeUtils.cacheRequestBodyAndRequest(exchange, + (serverHttpRequest) -> ServerRequest + .create(exchange.mutate().request(serverHttpRequest) + .build(), messageReaders) + .bodyToMono(inClass) + .doOnNext(objectValue -> exchange.getAttributes() + .put(CACHE_REQUEST_BODY_OBJECT_KEY, objectValue)) + .map(objectValue -> config.getPredicate() + .test(objectValue))); } }; } 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 d795193f..18fe8dc7 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 @@ -20,15 +20,24 @@ import java.net.URI; import java.util.HashMap; import java.util.LinkedHashSet; import java.util.Map; +import java.util.function.Function; import java.util.function.Predicate; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import org.springframework.cloud.gateway.filter.factory.GatewayFilterFactory; import org.springframework.cloud.gateway.handler.AsyncPredicate; +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.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.util.Assert; import org.springframework.web.server.ServerWebExchange; import org.springframework.web.util.UriComponentsBuilder; @@ -38,6 +47,8 @@ import org.springframework.web.util.UriComponentsBuilder; */ public final class ServerWebExchangeUtils { + private static final Log log = LogFactory.getLog(ServerWebExchangeUtils.class); + /** * Preserve-Host header attribute name. */ @@ -130,6 +141,18 @@ public final class ServerWebExchangeUtils { public static final String GATEWAY_ALREADY_PREFIXED_ATTR = qualify( "gatewayAlreadyPrefixed"); + /** + * Cached ServerHttpRequestDecorator attribute name. Used when + * {@link #cacheRequestBodyAndRequest(ServerWebExchange, Function)} is called. + */ + public static final String CACHED_SERVER_HTTP_REQUEST_DECORATOR_ATTR = "cachedServerHttpRequestDecorator"; + + /** + * Cached request body key. Used when {@link #cacheRequestBodyAndRequest(ServerWebExchange, Function)} + * or {@link #cacheRequestBody(ServerWebExchange, Function)} are called. + */ + public static final String CACHED_REQUEST_BODY_ATTR = "cachedRequestBody"; + private static final Log logger = LogFactory.getLog(ServerWebExchangeUtils.class); private ServerWebExchangeUtils() { @@ -247,4 +270,84 @@ public final class ServerWebExchangeUtils { new HashMap<>()); } + /** + * Caches the request body and the created {@link ServerHttpRequestDecorator} in + * ServerWebExchange attributes. Those attributes are {@link #CACHED_REQUEST_BODY_ATTR} + * and {@link #CACHED_SERVER_HTTP_REQUEST_DECORATOR_ATTR} respectively. This method + * is useful when the {@link ServerWebExchange} can not be modified, such as a + * {@link RoutePredicateFactory}. + * @param exchange the available ServerWebExchange. + * @param function a function that accepts the created ServerHttpRequestDecorator. + * @param generic type for the return {@link Mono}. + * @return Mono of type T created by the function parameter. + */ + public static Mono cacheRequestBodyAndRequest(ServerWebExchange exchange, + Function> function) { + return cacheRequestBody(exchange, true, function); + } + + /** + * Caches the request body in a ServerWebExchange attributes. The attribute is + * {@link #CACHED_REQUEST_BODY_ATTR}. This method is useful when the + * {@link ServerWebExchange} can be mutated, such as a {@link GatewayFilterFactory}/ + * @param exchange the available ServerWebExchange. + * @param function a function that accepts the created ServerHttpRequestDecorator. + * @param generic type for the return {@link Mono}. + * @return Mono of type T created by the function parameter. + */ + public static Mono cacheRequestBody(ServerWebExchange exchange, + Function> function) { + return cacheRequestBody(exchange, false, function); + } + + /** + * Caches the request body in a ServerWebExchange attribute. The attribute is + * {@link #CACHED_REQUEST_BODY_ATTR}. If this method is called from a location + * that can not mutate the ServerWebExchange (such as a Predicate), setting + * cacheDecoratedRequest to true will put a {@link ServerHttpRequestDecorator} in + * an attribute {@link #CACHED_SERVER_HTTP_REQUEST_DECORATOR_ATTR} for adaptation + * later. + * @param exchange the available ServerWebExchange. + * @param cacheDecoratedRequest if true, the ServerHttpRequestDecorator will be cached. + * @param function a function that accepts the created ServerHttpRequestDecorator. + * @param generic type for the return {@link Mono}. + * @return Mono of type T created by the function parameter. + */ + private static Mono cacheRequestBody(ServerWebExchange exchange, + boolean cacheDecoratedRequest, + Function> function) { + // Join all the DataBuffers so we have a single DataBuffer for the body + return DataBufferUtils.join(exchange.getRequest().getBody()) + .flatMap(dataBuffer -> { + if (dataBuffer.readableByteCount() > 0) { + if (log.isTraceEnabled()) { + log.trace("retaining body in exchange attribute"); + } + exchange.getAttributes().put(CACHED_REQUEST_BODY_ATTR, dataBuffer); + } + + ServerHttpRequestDecorator 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); + } + return function.apply(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 dcaa2e5f..f81d5b20 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 @@ -66,8 +66,8 @@ public class RetryGatewayFilterFactoryIntegrationTests extends BaseWebClientTest @Test public void retryFilterFailure() { - testClient.mutate().responseTimeout(Duration.ofSeconds(10)).build() - .get().uri("/retryalwaysfail?key=getjavafailure&count=4") + testClient.mutate().responseTimeout(Duration.ofSeconds(10)).build().get() + .uri("/retryalwaysfail?key=getjavafailure&count=4") .header(HttpHeaders.HOST, "www.retryjava.org").exchange().expectStatus() .is5xxServerError().expectBody(String.class).consumeWith(result -> { assertThat(result.getResponseBody()).contains("permanently broken"); diff --git a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/handler/predicate/ReadBodyPredicateFactoryTest.java b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/handler/predicate/ReadBodyPredicateFactoryTest.java index d849e482..8b202500 100644 --- a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/handler/predicate/ReadBodyPredicateFactoryTest.java +++ b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/handler/predicate/ReadBodyPredicateFactoryTest.java @@ -127,36 +127,36 @@ public class ReadBodyPredicateFactoryTest { } -} + static class Event { -class Event { + private String foo; - private String foo; + private String bar; - private String bar; + Event() { + } - Event() { - } + Event(String foo, String bar) { + this.foo = foo; + this.bar = bar; + } - Event(String foo, String bar) { - this.foo = foo; - this.bar = bar; - } + public String getFoo() { + return foo; + } - public String getFoo() { - return foo; - } + public void setFoo(String foo) { + this.foo = foo; + } - public void setFoo(String foo) { - this.foo = foo; - } + public String getBar() { + return bar; + } - public String getBar() { - return bar; - } + public void setBar(String bar) { + this.bar = bar; + } - public void setBar(String bar) { - this.bar = bar; } }