From a0508bbe1e25af7e81771fd27d1779ff7cc4bd9b Mon Sep 17 00:00:00 2001 From: Ryan Baxter Date: Wed, 15 Aug 2018 20:04:36 -0400 Subject: [PATCH 1/4] ReadBodyPredicate should set body to body of the original request, not the parsed object. --- .../config/GatewayAutoConfiguration.java | 4 +- .../predicate/ReadBodyPredicateFactory.java | 96 +++++++++++++------ 2 files changed, 68 insertions(+), 32 deletions(-) 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 7367455e..0ee16bc5 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 @@ -435,8 +435,8 @@ public class GatewayAutoConfiguration { } @Bean - public ReadBodyPredicateFactory readBodyPredicateFactory(ServerCodecConfigurer codecConfigurer) { - return new ReadBodyPredicateFactory(codecConfigurer); + public ReadBodyPredicateFactory readBodyPredicateFactory() { + return new ReadBodyPredicateFactory(); } @Bean 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 32ea6f9d..f4d32198 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 @@ -17,23 +17,31 @@ package org.springframework.cloud.gateway.handler.predicate; +import java.util.Collections; +import java.util.List; 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 org.springframework.cloud.gateway.support.BodyInserterContext; +import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import org.springframework.cloud.gateway.support.CachedBodyOutputMessage; import org.springframework.cloud.gateway.handler.AsyncPredicate; -import org.springframework.cloud.gateway.support.DefaultServerRequest; +import org.springframework.core.ResolvableType; +import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.http.HttpHeaders; +import org.springframework.http.ReactiveHttpInputMessage; +import org.springframework.http.codec.HttpMessageReader; import org.springframework.http.codec.ServerCodecConfigurer; import org.springframework.web.reactive.function.BodyInserter; import org.springframework.web.reactive.function.BodyInserters; -import org.springframework.web.reactive.function.server.ServerRequest; +import org.springframework.web.reactive.function.server.HandlerStrategies; import org.springframework.web.server.ServerWebExchange; import static org.springframework.cloud.gateway.filter.AdaptCachedBodyGlobalFilter.CACHED_REQUEST_BODY_KEY; @@ -47,11 +55,10 @@ public class ReadBodyPredicateFactory private static final String TEST_ATTRIBUTE = "read_body_predicate_test_attribute"; private static final String CACHE_REQUEST_BODY_OBJECT_KEY = "cachedRequestBodyObject"; - private final ServerCodecConfigurer codecConfigurer; + private static final List> messageReaders = HandlerStrategies.withDefaults().messageReaders(); - public ReadBodyPredicateFactory(ServerCodecConfigurer codecConfigurer) { + public ReadBodyPredicateFactory() { super(Config.class); - this.codecConfigurer = codecConfigurer; } @Override @@ -70,39 +77,48 @@ public class ReadBodyPredicateFactory try { boolean test = config.predicate.test(cachedBody); exchange.getAttributes().put(TEST_ATTRIBUTE, test); + return Mono.just(test); } catch(ClassCastException e) { if(LOGGER.isDebugEnabled()) { LOGGER.debug("Predicate test failed because class in predicate does not match the cached body object", e); } } - modifiedBody = Mono.just(cachedBody); + return Mono.just(false); } else { - ServerRequest serverRequest = new DefaultServerRequest(exchange); - // TODO: flux or mono - modifiedBody = serverRequest.bodyToMono(inClass) - // .log("modify_request_mono", Level.INFO) - .flatMap(body -> { - // TODO: migrate to async - exchange.getAttributes().put(CACHE_REQUEST_BODY_OBJECT_KEY, body); - boolean test = config.predicate.test(body); - exchange.getAttributes().put(TEST_ATTRIBUTE, test); - return Mono.just(body); - }); - } - BodyInserter bodyInserter = BodyInserters.fromPublisher(modifiedBody, inClass); - CachedBodyOutputMessage outputMessage = new CachedBodyOutputMessage(exchange, - exchange.getRequest().getHeaders()); - return bodyInserter.insert(outputMessage, new BodyInserterContext()) - // .log("modify_request", Level.INFO) - .then(Mono.defer(() -> { - boolean test = (Boolean) exchange.getAttributes() - .getOrDefault(TEST_ATTRIBUTE, Boolean.FALSE); - exchange.getAttributes().remove(TEST_ATTRIBUTE); - exchange.getAttributes().put(CACHED_REQUEST_BODY_KEY, + Flux origBody = exchange.getRequest().getBody().flatMap(dataBuffer -> { + ResolvableType type =ResolvableType.forClass(inClass); + for(HttpMessageReader messageReader: messageReaders) { + if(messageReader.canRead(type, exchange.getRequest().getHeaders().getContentType())) { + ReactiveHttpInputMessage inputMessage = new ReadBodyReactiveHttpInputMessage( + Flux.just(dataBuffer.factory().allocateBuffer().write(dataBuffer.asByteBuffer())), + exchange.getRequest().getHeaders()); + Function mapper = (bodyObj) -> { + exchange.getAttributes().put(CACHE_REQUEST_BODY_OBJECT_KEY, bodyObj); + boolean test = config.predicate.test(bodyObj); + exchange.getAttributes().put(TEST_ATTRIBUTE, test); + return Flux.just(dataBuffer.factory().allocateBuffer().write(dataBuffer.asByteBuffer())); + }; + return messageReader.read(type, inputMessage, Collections.EMPTY_MAP).flatMap(mapper); + } + } + return Flux.just(dataBuffer.factory().allocateBuffer().write(dataBuffer.asByteBuffer())); + }); + BodyInserter bodyInserter = BodyInserters.fromPublisher(origBody, DataBuffer.class); + CachedBodyOutputMessage outputMessage = new CachedBodyOutputMessage(exchange, + exchange.getRequest().getHeaders()); + return bodyInserter.insert(outputMessage, new BodyInserterContext()) + // .log("modify_request", Level.INFO) + .then(Mono.defer(() -> { + boolean test = (Boolean) exchange.getAttributes() + .getOrDefault(TEST_ATTRIBUTE, Boolean.FALSE); + exchange.getAttributes().remove(TEST_ATTRIBUTE); + exchange.getAttributes().put(CACHED_REQUEST_BODY_KEY, outputMessage.getBody()); - return Mono.just(test); - })); + return Mono.just(test); + })); + + } }; } @@ -113,6 +129,26 @@ public class ReadBodyPredicateFactory "ReadBodyPredicateFactory is only async."); } + static class ReadBodyReactiveHttpInputMessage implements ReactiveHttpInputMessage { + private Flux body; + private HttpHeaders httpHeaders; + + public ReadBodyReactiveHttpInputMessage(Flux body, HttpHeaders headers) { + this.body = body; + this.httpHeaders = headers; + } + + @Override + public Flux getBody() { + return body; + } + + @Override + public HttpHeaders getHeaders() { + return httpHeaders; + } + } + public static class Config { private Class inClass; private Predicate predicate; From 31ac60d19720308317b1492070b448831321a520 Mon Sep 17 00:00:00 2001 From: Ryan Baxter Date: Fri, 17 Aug 2018 20:14:19 -0400 Subject: [PATCH 2/4] Use a Collector to collect the request body into a map containing two copies. --- .../predicate/ReadBodyPredicateFactory.java | 124 +++++++++++++----- 1 file changed, 89 insertions(+), 35 deletions(-) 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 f4d32198..6ead30a2 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 @@ -18,10 +18,17 @@ package org.springframework.cloud.gateway.handler.predicate; import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.function.BiConsumer; +import java.util.function.BinaryOperator; import java.util.function.Function; import java.util.function.Predicate; +import java.util.function.Supplier; +import java.util.stream.Collector; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -37,7 +44,6 @@ import org.springframework.core.io.buffer.DataBuffer; import org.springframework.http.HttpHeaders; import org.springframework.http.ReactiveHttpInputMessage; import org.springframework.http.codec.HttpMessageReader; -import org.springframework.http.codec.ServerCodecConfigurer; import org.springframework.web.reactive.function.BodyInserter; import org.springframework.web.reactive.function.BodyInserters; @@ -45,6 +51,8 @@ import org.springframework.web.reactive.function.server.HandlerStrategies; import org.springframework.web.server.ServerWebExchange; import static org.springframework.cloud.gateway.filter.AdaptCachedBodyGlobalFilter.CACHED_REQUEST_BODY_KEY; +import static org.springframework.cloud.gateway.handler.predicate.ReadBodyPredicateFactory.DataBufferMapCollector.BODY_ONE; +import static org.springframework.cloud.gateway.handler.predicate.ReadBodyPredicateFactory.DataBufferMapCollector.BODY_TWO; /** * This predicate is BETA and may be subject to change in a future release. @@ -73,55 +81,101 @@ public class ReadBodyPredicateFactory // exception will be thrown. The below if/else caches the body object as a request attribute in the ServerWebExchange // so if this filter is run more than once (due to more than one route using it) we do not try to read the // request body multiple times - if(cachedBody != null) { + if (cachedBody != null) { try { boolean test = config.predicate.test(cachedBody); exchange.getAttributes().put(TEST_ATTRIBUTE, test); return Mono.just(test); - } catch(ClassCastException e) { - if(LOGGER.isDebugEnabled()) { + } catch (ClassCastException e) { + if (LOGGER.isDebugEnabled()) { LOGGER.debug("Predicate test failed because class in predicate does not match the cached body object", e); } } return Mono.just(false); } else { - Flux origBody = exchange.getRequest().getBody().flatMap(dataBuffer -> { - ResolvableType type =ResolvableType.forClass(inClass); - for(HttpMessageReader messageReader: messageReaders) { - if(messageReader.canRead(type, exchange.getRequest().getHeaders().getContentType())) { - ReactiveHttpInputMessage inputMessage = new ReadBodyReactiveHttpInputMessage( - Flux.just(dataBuffer.factory().allocateBuffer().write(dataBuffer.asByteBuffer())), - exchange.getRequest().getHeaders()); - Function mapper = (bodyObj) -> { - exchange.getAttributes().put(CACHE_REQUEST_BODY_OBJECT_KEY, bodyObj); - boolean test = config.predicate.test(bodyObj); - exchange.getAttributes().put(TEST_ATTRIBUTE, test); - return Flux.just(dataBuffer.factory().allocateBuffer().write(dataBuffer.asByteBuffer())); - }; - return messageReader.read(type, inputMessage, Collections.EMPTY_MAP).flatMap(mapper); - } - } - return Flux.just(dataBuffer.factory().allocateBuffer().write(dataBuffer.asByteBuffer())); - }); - BodyInserter bodyInserter = BodyInserters.fromPublisher(origBody, DataBuffer.class); - CachedBodyOutputMessage outputMessage = new CachedBodyOutputMessage(exchange, - exchange.getRequest().getHeaders()); - return bodyInserter.insert(outputMessage, new BodyInserterContext()) - // .log("modify_request", Level.INFO) - .then(Mono.defer(() -> { - boolean test = (Boolean) exchange.getAttributes() - .getOrDefault(TEST_ATTRIBUTE, Boolean.FALSE); - exchange.getAttributes().remove(TEST_ATTRIBUTE); - exchange.getAttributes().put(CACHED_REQUEST_BODY_KEY, - outputMessage.getBody()); - return Mono.just(test); - })); + return exchange.getRequest().getBody().collect(new DataBufferMapCollector()).flatMap(dataBufferMap -> { + BodyInserter bodyInserter = BodyInserters.fromPublisher(dataBufferMap.get(BODY_ONE), DataBuffer.class); + CachedBodyOutputMessage outputMessage = new CachedBodyOutputMessage(exchange, + exchange.getRequest().getHeaders()); + return bodyInserter.insert(outputMessage, new BodyInserterContext()) + // .log("modify_request", Level.INFO) + .then(Mono.defer(() -> { + ResolvableType type = ResolvableType.forClass(inClass); + for (HttpMessageReader messageReader : messageReaders) { + if (messageReader.canRead(type, exchange.getRequest().getHeaders().getContentType())) { + ReactiveHttpInputMessage inputMessage = new ReadBodyReactiveHttpInputMessage(dataBufferMap.get(BODY_TWO), + exchange.getRequest().getHeaders()); + Function mapper = (bodyObj) -> { + exchange.getAttributes().put(CACHE_REQUEST_BODY_OBJECT_KEY, bodyObj); + exchange.getAttributes().put(CACHED_REQUEST_BODY_KEY, + outputMessage.getBody()); + boolean test = config.predicate.test(bodyObj); + return Mono.just(test); + }; + return messageReader.readMono(type, inputMessage, Collections.EMPTY_MAP).flatMap(mapper); + } + } + return Mono.just(false); + })); + }); } }; } + /** + * This {@link Collector} is meant to collect the {@code Flux} from the request body into a {@link Map} + * which contains two copy of the body, one under the key {@code orig} and the other under the {@key copy}. + */ + class DataBufferMapCollector implements Collector>, Map>> { + public static final String BODY_ONE = "bodyOne"; + public static final String BODY_TWO = "bodyTwo"; + + @Override + public Supplier>> supplier() { + return () -> new HashMap>(); + } + + @Override + public BiConsumer>, DataBuffer> accumulator() { + return (dataBufferMap, dataBuffer) -> { + accumulate(BODY_ONE, dataBufferMap, dataBuffer); + accumulate(BODY_TWO, dataBufferMap, dataBuffer); + }; + } + + private void accumulate(String key, Map> dataBufferMap, DataBuffer dataBuffer) { + if (dataBufferMap.get(key) == null) { + dataBufferMap.put(key, Flux.just(copy(dataBuffer))); + } else { + dataBufferMap.put(key, dataBufferMap.get(key).mergeWith(Flux.just(copy(dataBuffer)))); + } + } + + @Override + public BinaryOperator>> combiner() { + return (map1, map2) -> { + map2.forEach((k, v) -> map1.merge(k, v, (v1, v2) -> v1.mergeWith(v2))); + return map1; + }; + } + + @Override + public Function>, Map>> finisher() { + return Function.identity(); + } + + @Override + public Set characteristics() { + return new HashSet(); + } + + private DataBuffer copy(DataBuffer dataBuffer) { + return dataBuffer.factory().allocateBuffer().write(dataBuffer.asByteBuffer()); + } + } + @Override @SuppressWarnings("unchecked") public Predicate apply(Config config) { From 23d075852ba5f3d41144ff6c790104ddd32a0660 Mon Sep 17 00:00:00 2001 From: Ryan Baxter Date: Fri, 17 Aug 2018 20:27:22 -0400 Subject: [PATCH 3/4] Adding Characteristics to the Collector --- .../gateway/handler/predicate/ReadBodyPredicateFactory.java | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) 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 6ead30a2..9218f1f0 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 @@ -17,6 +17,7 @@ package org.springframework.cloud.gateway.handler.predicate; +import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; @@ -131,6 +132,8 @@ public class ReadBodyPredicateFactory class DataBufferMapCollector implements Collector>, Map>> { public static final String BODY_ONE = "bodyOne"; public static final String BODY_TWO = "bodyTwo"; + private final Set CHARACTERISTICS = new HashSet<>(Arrays.asList( + Characteristics.IDENTITY_FINISH)); @Override public Supplier>> supplier() { @@ -168,7 +171,7 @@ public class ReadBodyPredicateFactory @Override public Set characteristics() { - return new HashSet(); + return CHARACTERISTICS; } private DataBuffer copy(DataBuffer dataBuffer) { From bdd970d562aa8b177bfe9d5ab674dddaa5987007 Mon Sep 17 00:00:00 2001 From: Ryan Baxter Date: Wed, 17 Oct 2018 11:17:46 -0400 Subject: [PATCH 4/4] Read the body twice by retaining it and using slice to give each read its own read/write indexes --- .../filter/AdaptCachedBodyGlobalFilter.java | 1 + .../predicate/ReadBodyPredicateFactory.java | 148 ++++-------------- 2 files changed, 31 insertions(+), 118 deletions(-) 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 16760f23..a7f758b2 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 @@ -39,6 +39,7 @@ public class AdaptCachedBodyGlobalFilter implements GlobalFilter, Ordered { return body; } }; + exchange.getAttributes().remove(CACHED_REQUEST_BODY_KEY); return chain.filter(exchange.mutate().request(decorator).build()); } 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 9218f1f0..05133916 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 @@ -17,43 +17,28 @@ package org.springframework.cloud.gateway.handler.predicate; -import java.util.Arrays; -import java.util.Collections; -import java.util.HashMap; -import java.util.HashSet; import java.util.List; import java.util.Map; -import java.util.Set; -import java.util.function.BiConsumer; -import java.util.function.BinaryOperator; -import java.util.function.Function; import java.util.function.Predicate; -import java.util.function.Supplier; -import java.util.stream.Collector; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.cloud.gateway.support.BodyInserterContext; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import org.springframework.cloud.gateway.support.CachedBodyOutputMessage; import org.springframework.cloud.gateway.handler.AsyncPredicate; -import org.springframework.core.ResolvableType; import org.springframework.core.io.buffer.DataBuffer; -import org.springframework.http.HttpHeaders; -import org.springframework.http.ReactiveHttpInputMessage; +import org.springframework.core.io.buffer.DataBufferUtils; import org.springframework.http.codec.HttpMessageReader; -import org.springframework.web.reactive.function.BodyInserter; -import org.springframework.web.reactive.function.BodyInserters; +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; -import static org.springframework.cloud.gateway.handler.predicate.ReadBodyPredicateFactory.DataBufferMapCollector.BODY_ONE; -import static org.springframework.cloud.gateway.handler.predicate.ReadBodyPredicateFactory.DataBufferMapCollector.BODY_TWO; /** * This predicate is BETA and may be subject to change in a future release. @@ -95,90 +80,37 @@ public class ReadBodyPredicateFactory } return Mono.just(false); } else { - return exchange.getRequest().getBody().collect(new DataBufferMapCollector()).flatMap(dataBufferMap -> { - BodyInserter bodyInserter = BodyInserters.fromPublisher(dataBufferMap.get(BODY_ONE), DataBuffer.class); - CachedBodyOutputMessage outputMessage = new CachedBodyOutputMessage(exchange, - exchange.getRequest().getHeaders()); - return bodyInserter.insert(outputMessage, new BodyInserterContext()) - // .log("modify_request", Level.INFO) - .then(Mono.defer(() -> { - ResolvableType type = ResolvableType.forClass(inClass); - for (HttpMessageReader messageReader : messageReaders) { - if (messageReader.canRead(type, exchange.getRequest().getHeaders().getContentType())) { - ReactiveHttpInputMessage inputMessage = new ReadBodyReactiveHttpInputMessage(dataBufferMap.get(BODY_TWO), - exchange.getRequest().getHeaders()); - Function mapper = (bodyObj) -> { - exchange.getAttributes().put(CACHE_REQUEST_BODY_OBJECT_KEY, bodyObj); - exchange.getAttributes().put(CACHED_REQUEST_BODY_KEY, - outputMessage.getBody()); - boolean test = config.predicate.test(bodyObj); - return Mono.just(test); - }; - return messageReader.readMono(type, inputMessage, Collections.EMPTY_MAP).flatMap(mapper); - } - } - return Mono.just(false); - })); + //Join all the DataBuffers so we have a single DataBuffer for the body + return DataBufferUtils.join(exchange.getRequest().getBody()) + .flatMap(dataBuffer -> { + //Update the retain counts so we can read the body twice, once to parse into an object + //that we can test the predicate against and a second time when the HTTP client sends + //the request downstream + //Note: if we end up reading the body twice we will run into a problem, but as of right + //now there is no good use case for doing this + DataBufferUtils.retain(dataBuffer); + //Make a slice for each read so each read has its own read/write indexes + Flux cachedFlux = Flux.defer(() -> Flux.just(dataBuffer.slice(0, dataBuffer.readableByteCount()))); + + 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)); + }); - }); } }; } - /** - * This {@link Collector} is meant to collect the {@code Flux} from the request body into a {@link Map} - * which contains two copy of the body, one under the key {@code orig} and the other under the {@key copy}. - */ - class DataBufferMapCollector implements Collector>, Map>> { - public static final String BODY_ONE = "bodyOne"; - public static final String BODY_TWO = "bodyTwo"; - private final Set CHARACTERISTICS = new HashSet<>(Arrays.asList( - Characteristics.IDENTITY_FINISH)); - - @Override - public Supplier>> supplier() { - return () -> new HashMap>(); - } - - @Override - public BiConsumer>, DataBuffer> accumulator() { - return (dataBufferMap, dataBuffer) -> { - accumulate(BODY_ONE, dataBufferMap, dataBuffer); - accumulate(BODY_TWO, dataBufferMap, dataBuffer); - }; - } - - private void accumulate(String key, Map> dataBufferMap, DataBuffer dataBuffer) { - if (dataBufferMap.get(key) == null) { - dataBufferMap.put(key, Flux.just(copy(dataBuffer))); - } else { - dataBufferMap.put(key, dataBufferMap.get(key).mergeWith(Flux.just(copy(dataBuffer)))); - } - } - - @Override - public BinaryOperator>> combiner() { - return (map1, map2) -> { - map2.forEach((k, v) -> map1.merge(k, v, (v1, v2) -> v1.mergeWith(v2))); - return map1; - }; - } - - @Override - public Function>, Map>> finisher() { - return Function.identity(); - } - - @Override - public Set characteristics() { - return CHARACTERISTICS; - } - - private DataBuffer copy(DataBuffer dataBuffer) { - return dataBuffer.factory().allocateBuffer().write(dataBuffer.asByteBuffer()); - } - } - @Override @SuppressWarnings("unchecked") public Predicate apply(Config config) { @@ -186,26 +118,6 @@ public class ReadBodyPredicateFactory "ReadBodyPredicateFactory is only async."); } - static class ReadBodyReactiveHttpInputMessage implements ReactiveHttpInputMessage { - private Flux body; - private HttpHeaders httpHeaders; - - public ReadBodyReactiveHttpInputMessage(Flux body, HttpHeaders headers) { - this.body = body; - this.httpHeaders = headers; - } - - @Override - public Flux getBody() { - return body; - } - - @Override - public HttpHeaders getHeaders() { - return httpHeaders; - } - } - public static class Config { private Class inClass; private Predicate predicate;