SpringCloud/spring-cloud-gateway
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

ReadBodyPredicate should set body to body of the original request, not the parsed object.

Browse Source
This commit is contained in:
Ryan Baxter
2018-08-15 20:04:36 -04:00
parent cf28327eeb
commit a0508bbe1e
2 changed files with 68 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/config/GatewayAutoConfiguration.java
View File

@@ -435,8 +435,8 @@ public class GatewayAutoConfiguration {
}
@Bean
public ReadBodyPredicateFactory readBodyPredicateFactory(ServerCodecConfigurer codecConfigurer) {
return new ReadBodyPredicateFactory(codecConfigurer);
public ReadBodyPredicateFactory readBodyPredicateFactory() {
return new ReadBodyPredicateFactory();
}
@Bean

96
spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/handler/predicate/ReadBodyPredicateFactory.java
View File

@@ -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<HttpMessageReader<?>> 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<DataBuffer> 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<DataBuffer> body;
private HttpHeaders httpHeaders;
public ReadBodyReactiveHttpInputMessage(Flux<DataBuffer> body, HttpHeaders headers) {
this.body = body;
this.httpHeaders = headers;
}
@Override
public Flux<DataBuffer> getBody() {
return body;
}
@Override
public HttpHeaders getHeaders() {
return httpHeaders;
}
}
public static class Config {
private Class inClass;
private Predicate predicate;