Read write body filters and predicate (#278)

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

@@ -34,6 +34,7 @@ import org.springframework.boot.autoconfigure.web.reactive.HttpHandlerAutoConfig
 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.ForwardRoutingFilter;
 import org.springframework.cloud.gateway.filter.GlobalFilter;
 import org.springframework.cloud.gateway.filter.WeightCalculatorWebFilter;
@@ -62,6 +63,8 @@ import org.springframework.cloud.gateway.filter.factory.SetRequestHeaderGatewayF
 import org.springframework.cloud.gateway.filter.factory.SetResponseHeaderGatewayFilterFactory;
 import org.springframework.cloud.gateway.filter.factory.SetStatusGatewayFilterFactory;
 import org.springframework.cloud.gateway.filter.factory.StripPrefixGatewayFilterFactory;
+import org.springframework.cloud.gateway.filter.factory.rewrite.ModifyRequestBodyGatewayFilterFactory;
+import org.springframework.cloud.gateway.filter.factory.rewrite.ModifyResponseBodyGatewayFilterFactory;
 import org.springframework.cloud.gateway.filter.headers.ForwardedHeadersFilter;
 import org.springframework.cloud.gateway.filter.headers.HttpHeadersFilter;
 import org.springframework.cloud.gateway.filter.headers.RemoveHopByHopHeadersFilter;
@@ -80,6 +83,7 @@ import org.springframework.cloud.gateway.handler.predicate.HostRoutePredicateFac
 import org.springframework.cloud.gateway.handler.predicate.MethodRoutePredicateFactory;
 import org.springframework.cloud.gateway.handler.predicate.PathRoutePredicateFactory;
 import org.springframework.cloud.gateway.handler.predicate.QueryRoutePredicateFactory;
+import org.springframework.cloud.gateway.handler.predicate.ReadBodyPredicateFactory;
 import org.springframework.cloud.gateway.handler.predicate.RemoteAddrRoutePredicateFactory;
 import org.springframework.cloud.gateway.handler.predicate.RoutePredicateFactory;
 import org.springframework.cloud.gateway.handler.predicate.WeightRoutePredicateFactory;
@@ -100,6 +104,7 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.DependsOn;
 import org.springframework.context.annotation.Primary;
+import org.springframework.http.codec.ServerCodecConfigurer;
 import org.springframework.util.StringUtils;
 import org.springframework.validation.Validator;
 import org.springframework.web.reactive.DispatcherHandler;
@@ -307,6 +312,12 @@ public class GatewayAutoConfiguration {
 
 
 	// GlobalFilter beans
+
+	@Bean
+	public AdaptCachedBodyGlobalFilter adaptCachedBodyGlobalFilter() {
+		return new AdaptCachedBodyGlobalFilter();
+	}
+
 	@Bean
 	public RouteToRequestUrlFilter routeToRequestUrlFilter() {
 		return new RouteToRequestUrlFilter();
@@ -394,6 +405,11 @@ public class GatewayAutoConfiguration {
 		return new QueryRoutePredicateFactory();
 	}
 
+	@Bean
+	public ReadBodyPredicateFactory readBodyPredicateFactory(ServerCodecConfigurer codecConfigurer) {
+		return new ReadBodyPredicateFactory(codecConfigurer);
+	}
+
 	@Bean
 	public RemoteAddrRoutePredicateFactory remoteAddrRoutePredicateFactory() {
 		return new RemoteAddrRoutePredicateFactory();
@@ -431,6 +447,16 @@ public class GatewayAutoConfiguration {
 		}
 	}
 
+	@Bean
+	public ModifyRequestBodyGatewayFilterFactory modifyRequestBodyGatewayFilterFactory(ServerCodecConfigurer codecConfigurer) {
+		return new ModifyRequestBodyGatewayFilterFactory(codecConfigurer);
+	}
+
+	@Bean
+	public ModifyResponseBodyGatewayFilterFactory modifyResponseBodyGatewayFilterFactory(ServerCodecConfigurer codecConfigurer) {
+		return new ModifyResponseBodyGatewayFilterFactory(codecConfigurer);
+	}
+
 	@Bean
 	public PrefixPathGatewayFilterFactory prefixPathGatewayFilterFactory() {
 		return new PrefixPathGatewayFilterFactory();

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/AdaptCachedBodyGlobalFilter.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2013-2018 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
+ *
+ *      http://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 org.springframework.core.Ordered;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.http.server.reactive.ServerHttpRequestDecorator;
+import org.springframework.web.server.ServerWebExchange;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public class AdaptCachedBodyGlobalFilter implements GlobalFilter, Ordered {
+
+	public static final String CACHED_REQUEST_BODY_KEY = "cachedRequestBody";
+
+	@Override
+    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
+
+        Flux<DataBuffer> body = exchange.getAttributeOrDefault(CACHED_REQUEST_BODY_KEY, null);
+        if (body != null) {
+            ServerHttpRequestDecorator decorator = new ServerHttpRequestDecorator(exchange.getRequest()) {
+                @Override
+                public Flux<DataBuffer> getBody() {
+                    return body;
+                }
+            };
+            return chain.filter(exchange.mutate().request(decorator).build());
+        }
+
+        return chain.filter(exchange);
+    }
+
+    @Override
+    public int getOrder() {
+        return Ordered.LOWEST_PRECEDENCE - 10; //probably needs to change if combined with other filters that modify request body
+    }
+}

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/rewrite/HttpMessageWriterResponse.java

@@ -0,0 +1,109 @@
+/*
+ * Copyright 2013-2018 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
+ *
+ *      http://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.factory.rewrite;
+
+import org.reactivestreams.Publisher;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseCookie;
+import org.springframework.http.server.reactive.ServerHttpResponse;
+import org.springframework.util.MultiValueMap;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.function.Supplier;
+
+/**
+ * Response who's job it is to gather the Publisher<DataBuffer> from the writeWith message
+ * during a call to HttpMessageWriter.write. Also gathers any headers set there.
+ */
+public class HttpMessageWriterResponse implements ServerHttpResponse {
+
+	private final HttpHeaders headers = new HttpHeaders();
+	private final DataBufferFactory dataBufferFactory;
+
+	private Publisher<? extends DataBuffer> body;
+
+    public HttpMessageWriterResponse(DataBufferFactory dataBufferFactory) {
+		this.dataBufferFactory = dataBufferFactory;
+	}
+
+	@Override
+	public HttpHeaders getHeaders() {
+		return this.headers;
+	}
+
+	@Override
+    public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
+        this.body = body;
+        return Mono.empty();
+    }
+
+    @Override
+    public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> body) {
+    	//TODO: is this kosher?
+		return writeWith(Flux.from(body)
+						.flatMapSequential(p -> p));
+    }
+
+    public Publisher<? extends DataBuffer> getBody() {
+        return body;
+    }
+
+	@Override
+	public boolean setStatusCode(HttpStatus status) {
+		return false;
+	}
+
+	@Override
+	public HttpStatus getStatusCode() {
+		return null;
+	}
+
+	@Override
+	public MultiValueMap<String, ResponseCookie> getCookies() {
+		return null;
+	}
+
+	@Override
+	public void addCookie(ResponseCookie cookie) {
+
+	}
+
+	@Override
+	public DataBufferFactory bufferFactory() {
+		return this.dataBufferFactory;
+	}
+
+	@Override
+	public void beforeCommit(Supplier<? extends Mono<Void>> action) {
+
+	}
+
+	@Override
+	public boolean isCommitted() {
+		return false;
+	}
+
+	@Override
+	public Mono<Void> setComplete() {
+		return null;
+	}
+}

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/rewrite/ModifyRequestBodyGatewayFilterFactory.java

@@ -0,0 +1,150 @@
+/*
+ * Copyright 2013-2018 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
+ *
+ *      http://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.factory.rewrite;
+
+import java.util.Optional;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.gateway.filter.GatewayFilter;
+import org.springframework.cloud.gateway.filter.factory.AbstractGatewayFilterFactory;
+import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.http.codec.HttpMessageReader;
+import org.springframework.http.codec.HttpMessageWriter;
+import org.springframework.http.codec.ServerCodecConfigurer;
+import org.springframework.http.server.reactive.ServerHttpRequestDecorator;
+
+import static org.springframework.cloud.gateway.filter.factory.rewrite.RewriteUtils.process;
+
+public class ModifyRequestBodyGatewayFilterFactory
+		extends AbstractGatewayFilterFactory<ModifyRequestBodyGatewayFilterFactory.Config> {
+
+	private final ServerCodecConfigurer codecConfigurer;
+
+	public ModifyRequestBodyGatewayFilterFactory(ServerCodecConfigurer codecConfigurer) {
+		super(Config.class);
+		this.codecConfigurer = codecConfigurer;
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	public GatewayFilter apply(Config config) {
+		return (exchange, chain) -> {
+			Class inClass = config.getInClass();
+
+			MediaType mediaType = exchange.getRequest().getHeaders().getContentType();
+			ResolvableType inElementType = ResolvableType.forClass(inClass);
+			Optional<HttpMessageReader<?>> reader = RewriteUtils.getHttpMessageReader(codecConfigurer, inElementType, mediaType);
+
+			if (reader.isPresent()) {
+				Mono<Object> readMono = reader.get()
+						.readMono(inElementType, exchange.getRequest(), null)
+						.cast(Object.class);
+
+				return process(readMono, peek -> {
+					ResolvableType outElementType = ResolvableType
+							.forClass(config.getOutClass());
+					Optional<HttpMessageWriter<?>> writer = RewriteUtils.getHttpMessageWriter(codecConfigurer, outElementType, mediaType);
+
+					if (writer.isPresent()) {
+						Object data = config.rewriteFunction.apply(exchange, peek);
+
+						//TODO: deal with multivalue? ie Flux
+						Publisher publisher = Mono.just(data);
+
+						HttpMessageWriterResponse fakeResponse = new HttpMessageWriterResponse(exchange.getResponse().bufferFactory());
+						writer.get().write(publisher, inElementType, mediaType,
+								fakeResponse, null);
+						ServerHttpRequestDecorator decorator = new ServerHttpRequestDecorator(
+								exchange.getRequest()) {
+							@Override
+							public HttpHeaders getHeaders() {
+								HttpHeaders httpHeaders = new HttpHeaders();
+								httpHeaders.putAll(super.getHeaders());
+								// TODO: this causes a 'HTTP/1.1 411 Length Required' on
+								// httpbin.org
+								httpHeaders.set(HttpHeaders.TRANSFER_ENCODING, "chunked");
+								if (fakeResponse.getHeaders().getContentType() != null) {
+									httpHeaders.setContentType(
+											fakeResponse.getHeaders().getContentType());
+								}
+								return httpHeaders;
+							}
+
+							@Override
+							public Flux<DataBuffer> getBody() {
+								return (Flux<DataBuffer>) fakeResponse.getBody();
+							}
+						};
+						return chain.filter(exchange.mutate().request(decorator).build());
+					}
+					return chain.filter(exchange);
+				});
+
+			}
+			return chain.filter(exchange);
+		};
+	}
+
+	public static class Config {
+		private Class inClass;
+		private Class outClass;
+
+		private RewriteFunction rewriteFunction;
+
+		public Class getInClass() {
+			return inClass;
+		}
+
+		public Config setInClass(Class inClass) {
+			this.inClass = inClass;
+			return this;
+		}
+
+		public Class getOutClass() {
+			return outClass;
+		}
+
+		public Config setOutClass(Class outClass) {
+			this.outClass = outClass;
+			return this;
+		}
+
+		public RewriteFunction getRewriteFunction() {
+			return rewriteFunction;
+		}
+
+		public <T, R> Config setRewriteFunction(Class<T> inClass, Class<R> outClass,
+				RewriteFunction<T, R> rewriteFunction) {
+			setInClass(inClass);
+			setOutClass(outClass);
+			setRewriteFunction(rewriteFunction);
+			return this;
+		}
+
+		public Config setRewriteFunction(RewriteFunction rewriteFunction) {
+			this.rewriteFunction = rewriteFunction;
+			return this;
+		}
+	}
+}

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/rewrite/ModifyResponseBodyGatewayFilterFactory.java

@@ -0,0 +1,182 @@
+/*
+ * Copyright 2013-2018 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
+ *
+ *      http://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.factory.rewrite;
+
+import java.util.Optional;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.gateway.filter.GatewayFilter;
+import org.springframework.cloud.gateway.filter.GatewayFilterChain;
+import org.springframework.cloud.gateway.filter.NettyWriteResponseFilter;
+import org.springframework.cloud.gateway.filter.factory.AbstractGatewayFilterFactory;
+import org.springframework.core.Ordered;
+import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.http.ReactiveHttpInputMessage;
+import org.springframework.http.codec.HttpMessageReader;
+import org.springframework.http.codec.HttpMessageWriter;
+import org.springframework.http.codec.ServerCodecConfigurer;
+import org.springframework.http.server.reactive.ServerHttpResponseDecorator;
+import org.springframework.web.server.ServerWebExchange;
+
+import static org.springframework.cloud.gateway.filter.factory.rewrite.RewriteUtils.getHttpMessageReader;
+import static org.springframework.cloud.gateway.filter.factory.rewrite.RewriteUtils.getHttpMessageWriter;
+
+public class ModifyResponseBodyGatewayFilterFactory
+		extends AbstractGatewayFilterFactory<ModifyResponseBodyGatewayFilterFactory.Config> {
+
+	private final ServerCodecConfigurer codecConfigurer;
+
+	public ModifyResponseBodyGatewayFilterFactory(ServerCodecConfigurer codecConfigurer) {
+		super(Config.class);
+		this.codecConfigurer = codecConfigurer;
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	public GatewayFilter apply(Config config) {
+		return new ModifyResponseGatewayFilter(config);
+	}
+
+	public class ModifyResponseGatewayFilter implements GatewayFilter, Ordered {
+		private final Config config;
+
+		public ModifyResponseGatewayFilter(Config config) {
+			this.config = config;
+		}
+
+		@Override
+		public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
+			ServerHttpResponseDecorator responseDecorator = new ServerHttpResponseDecorator(exchange.getResponse()) {
+				@Override
+				public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
+
+					ResolvableType inElementType = ResolvableType.forClass(config.getInClass());
+					ResolvableType outElementType = ResolvableType.forClass(config.getOutClass());
+					MediaType contentType = exchange.getResponse().getHeaders().getContentType();
+					Optional<HttpMessageReader<?>> reader = getHttpMessageReader(codecConfigurer, inElementType, contentType);
+					Optional<HttpMessageWriter<?>> writer = getHttpMessageWriter(codecConfigurer, outElementType, null);
+
+					if (reader.isPresent() && writer.isPresent()) {
+
+						ResponseAdapter responseAdapter = new ResponseAdapter(body, getDelegate().getHeaders());
+
+						Flux<?> modified = reader.get().read(inElementType, responseAdapter, null)
+								.cast(inElementType.resolve())
+								.flatMap(originalBody -> Flux.just(config.rewriteFunction.apply(exchange, originalBody)))
+								.cast(outElementType.resolve());
+
+						return getDelegate().writeWith(
+								writer.get().write((Publisher)modified, outElementType, null, getDelegate(), null)
+						);
+
+					}
+					// TODO: error? log?
+
+					return getDelegate().writeWith(body);
+				}
+
+				@Override
+				public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> body) {
+					return writeWith(Flux.from(body)
+							.flatMapSequential(p -> p));
+				}
+			};
+
+			return chain.filter(exchange.mutate().response(responseDecorator).build());
+		}
+
+		@Override
+		public int getOrder() {
+			return NettyWriteResponseFilter.WRITE_RESPONSE_FILTER_ORDER - 1;
+		}
+
+	}
+
+	public class ResponseAdapter implements ReactiveHttpInputMessage {
+
+		private final Flux<DataBuffer> flux;
+		private final HttpHeaders headers;
+
+		public ResponseAdapter(Publisher<? extends DataBuffer> body, HttpHeaders headers) {
+			this.headers = headers;
+			if (body instanceof Flux) {
+				flux = (Flux) body;
+			} else {
+				flux = ((Mono)body).flux();
+			}
+		}
+
+		@Override
+		public Flux<DataBuffer> getBody() {
+			return flux;
+		}
+
+		@Override
+		public HttpHeaders getHeaders() {
+			return headers;
+		}
+	}
+
+	public static class Config {
+		private Class inClass;
+		private Class outClass;
+
+		private RewriteFunction rewriteFunction;
+
+		public Class getInClass() {
+			return inClass;
+		}
+
+		public Config setInClass(Class inClass) {
+			this.inClass = inClass;
+			return this;
+		}
+
+		public Class getOutClass() {
+			return outClass;
+		}
+
+		public Config setOutClass(Class outClass) {
+			this.outClass = outClass;
+			return this;
+		}
+
+		public RewriteFunction getRewriteFunction() {
+			return rewriteFunction;
+		}
+
+		public <T, R> Config setRewriteFunction(Class<T> inClass, Class<R> outClass,
+				RewriteFunction<T, R> rewriteFunction) {
+			setInClass(inClass);
+			setOutClass(outClass);
+			setRewriteFunction(rewriteFunction);
+			return this;
+		}
+
+		public Config setRewriteFunction(RewriteFunction rewriteFunction) {
+			this.rewriteFunction = rewriteFunction;
+			return this;
+		}
+	}
+}

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/rewrite/RewriteFunction.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2013-2018 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
+ *
+ *      http://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.factory.rewrite;
+
+import org.springframework.web.server.ServerWebExchange;
+
+import java.util.function.BiFunction;
+
+public interface RewriteFunction<T, R> extends BiFunction<ServerWebExchange, T, R> {
+}

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/rewrite/RewriteUtils.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2013-2018 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
+ *
+ *      http://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.factory.rewrite;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Function;
+
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.MonoProcessor;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.http.MediaType;
+import org.springframework.http.codec.CodecConfigurer;
+import org.springframework.http.codec.HttpMessageReader;
+import org.springframework.http.codec.HttpMessageWriter;
+
+public abstract class RewriteUtils {
+
+	public static <T, R> R process(Mono<T> mono, Function<T, R> consumer) {
+		MonoProcessor<T> processor = MonoProcessor.create();
+		mono.subscribeWith(processor);
+		if (processor.isTerminated()) {
+			Throwable error = processor.getError();
+			if (error != null) {
+				throw (RuntimeException) error;
+			}
+			T peek = processor.peek();
+
+			return consumer.apply(peek);
+		}
+		else {
+			// Should never happen...
+			throw new IllegalStateException(
+					"SyncInvocableHandlerMethod should have completed synchronously.");
+		}
+	}
+
+	public static Optional<HttpMessageReader<?>> getHttpMessageReader(CodecConfigurer codecConfigurer, ResolvableType inElementType, MediaType mediaType) {
+		List<HttpMessageReader<?>> readers = codecConfigurer.getReaders();
+		return readers.stream()
+                .filter(r -> r.canRead(inElementType, mediaType))
+				.findFirst();
+	}
+
+	public static Optional<HttpMessageWriter<?>> getHttpMessageWriter(CodecConfigurer codecConfigurer, ResolvableType outElementType, MediaType mediaType) {
+		return codecConfigurer
+                                .getWriters().stream()
+                                .filter(w -> w.canWrite(outElementType, mediaType))
+                                .findFirst();
+	}
+}

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

@@ -0,0 +1,105 @@
+/*
+ * Copyright 2013-2018 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
+ *
+ *      http://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.handler.predicate;
+
+import java.util.Optional;
+import java.util.function.Predicate;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.gateway.filter.factory.rewrite.HttpMessageWriterResponse;
+import org.springframework.core.ResolvableType;
+import org.springframework.http.MediaType;
+import org.springframework.http.codec.HttpMessageReader;
+import org.springframework.http.codec.HttpMessageWriter;
+import org.springframework.http.codec.ServerCodecConfigurer;
+import org.springframework.web.server.ServerWebExchange;
+
+import static org.springframework.cloud.gateway.filter.AdaptCachedBodyGlobalFilter.CACHED_REQUEST_BODY_KEY;
+import static org.springframework.cloud.gateway.filter.factory.rewrite.RewriteUtils.getHttpMessageReader;
+import static org.springframework.cloud.gateway.filter.factory.rewrite.RewriteUtils.getHttpMessageWriter;
+import static org.springframework.cloud.gateway.filter.factory.rewrite.RewriteUtils.process;
+
+public class ReadBodyPredicateFactory extends AbstractRoutePredicateFactory<ReadBodyPredicateFactory.Config> {
+
+	private final ServerCodecConfigurer codecConfigurer;
+
+    public ReadBodyPredicateFactory(ServerCodecConfigurer codecConfigurer) {
+        super(Config.class);
+		this.codecConfigurer = codecConfigurer;
+	}
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public Predicate<ServerWebExchange> apply(Config config) {
+        return exchange -> {
+            MediaType mediaType = exchange.getRequest().getHeaders().getContentType();
+            ResolvableType elementType = ResolvableType.forClass(config.getInClass());
+            Optional<HttpMessageReader<?>> reader = getHttpMessageReader(codecConfigurer, elementType, mediaType);
+            boolean answer = false;
+if (reader.isPresent()) {
+                Mono<Object> readMono = reader.get()
+                        .readMono(elementType, exchange.getRequest(), null)
+                        .cast(Object.class);
+                answer = process(readMono, peek -> {
+                    Optional<HttpMessageWriter<?>> writer = getHttpMessageWriter(codecConfigurer, elementType, mediaType);
+
+                    if (writer.isPresent()) {
+                        Publisher publisher = Mono.just(peek);
+                        HttpMessageWriterResponse fakeResponse = new HttpMessageWriterResponse(exchange.getResponse().bufferFactory());
+                        writer.get().write(publisher, elementType, mediaType, fakeResponse, null);
+                        exchange.getAttributes().put(CACHED_REQUEST_BODY_KEY, fakeResponse.getBody());
+                    }
+                    return config.getPredicate().test(peek);
+                });
+
+            }
+            return answer;
+        };
+    }
+
+    public static class Config {
+        private Class inClass;
+        private Predicate predicate;
+
+        public Class getInClass() {
+            return inClass;
+        }
+
+        public Config setInClass(Class inClass) {
+            this.inClass = inClass;
+            return this;
+        }
+
+        public Predicate getPredicate() {
+            return predicate;
+        }
+
+        public <T> Config setPredicate(Class<T> inClass, Predicate<T> predicate) {
+            setInClass(inClass);
+            this.predicate = predicate;
+            return this;
+        }
+
+        public Config setPredicate(Predicate predicate) {
+            this.predicate = predicate;
+            return this;
+        }
+    }
+}

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/route/builder/GatewayFilterSpec.java

@@ -45,6 +45,9 @@ import org.springframework.cloud.gateway.filter.factory.SetRequestHeaderGatewayF
 import org.springframework.cloud.gateway.filter.factory.SetResponseHeaderGatewayFilterFactory;
 import org.springframework.cloud.gateway.filter.factory.SetStatusGatewayFilterFactory;
 import org.springframework.cloud.gateway.filter.factory.StripPrefixGatewayFilterFactory;
+import org.springframework.cloud.gateway.filter.factory.rewrite.ModifyRequestBodyGatewayFilterFactory;
+import org.springframework.cloud.gateway.filter.factory.rewrite.ModifyResponseBodyGatewayFilterFactory;
+import org.springframework.cloud.gateway.filter.factory.rewrite.RewriteFunction;
 import org.springframework.cloud.gateway.filter.ratelimit.RateLimiter;
 import org.springframework.cloud.gateway.route.Route;
 import org.springframework.core.Ordered;
@@ -116,6 +119,16 @@ public class GatewayFilterSpec extends UriSpec {
 		return filter(factory.apply(configConsumer));
 	}
 
+	public <T, R> GatewayFilterSpec modifyRequestBody(Class<T> inClass, Class<R> outClass, RewriteFunction<T, R> rewriteFunction) {
+		return filter(getBean(ModifyRequestBodyGatewayFilterFactory.class)
+				.apply(c -> c.setRewriteFunction(inClass, outClass, rewriteFunction)));
+	}
+
+	public <T, R> GatewayFilterSpec modifyResponseBody(Class<T> inClass, Class<R> outClass, RewriteFunction<T, R> rewriteFunction) {
+		return filter(getBean(ModifyResponseBodyGatewayFilterFactory.class)
+				.apply(c -> c.setRewriteFunction(inClass, outClass, rewriteFunction)));
+	}
+
 	public GatewayFilterSpec prefixPath(String prefix) {
 		return filter(getBean(PrefixPathGatewayFilterFactory.class)
 				.apply(c -> c.setPrefix(prefix)));

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/route/builder/PredicateSpec.java

@@ -28,6 +28,7 @@ import org.springframework.cloud.gateway.handler.predicate.HostRoutePredicateFac
 import org.springframework.cloud.gateway.handler.predicate.MethodRoutePredicateFactory;
 import org.springframework.cloud.gateway.handler.predicate.PathRoutePredicateFactory;
 import org.springframework.cloud.gateway.handler.predicate.QueryRoutePredicateFactory;
+import org.springframework.cloud.gateway.handler.predicate.ReadBodyPredicateFactory;
 import org.springframework.cloud.gateway.handler.predicate.RemoteAddrRoutePredicateFactory;
 import org.springframework.cloud.gateway.handler.predicate.WeightRoutePredicateFactory;
 import org.springframework.cloud.gateway.route.Route;
@@ -104,6 +105,11 @@ public class PredicateSpec extends UriSpec {
 				.apply(c -> c.setPattern(pattern)));
 	}
 
+	public <T> BooleanSpec readBody(Class<T> inClass, Predicate<T> predicate) {
+		return predicate(getBean(ReadBodyPredicateFactory.class)
+				.apply(c -> c.setPredicate(inClass, predicate)));
+	}
+
 	public BooleanSpec query(String param, String regex) {
 		return predicate(getBean(QueryRoutePredicateFactory.class)
 				.apply(c -> c.setParam(param).setRegexp(regex)));

spring-cloud-gateway-sample/src/main/java/org/springframework/cloud/gateway/sample/GatewaySampleApplication.java

@@ -17,8 +17,10 @@
 
 package org.springframework.cloud.gateway.sample;
 
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.SpringBootConfiguration;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
@@ -40,19 +42,64 @@ import org.springframework.web.reactive.function.server.ServerResponse;
 @Import(AdditionalRoutes.class)
 public class GatewaySampleApplication {
 
+	@Value("${route.uri:http://httpbin.org:80}")
+	String uri;
+
 	@Bean
 	public RouteLocator customRouteLocator(RouteLocatorBuilder builder) {
 		//@formatter:off
+		// String uri = "http://httpbin.org:80";
+		// String uri = "http://localhost:9080";
 		return builder.routes()
 				.route(r -> r.host("**.abc.org").and().path("/image/png")
 					.filters(f ->
 							f.addResponseHeader("X-TestHeader", "foobar"))
-					.uri("http://httpbin.org:80")
+					.uri(uri)
+				)
+				.route("read_body_pred", r -> r.host("*.readbody.org")
+						.and().readBody(String.class,
+										s -> s.trim().equalsIgnoreCase("hello"))
+					.filters(f ->
+							f.addRequestHeader("X-TestHeader", "read_body_pred")
+					).uri(uri)
+				)
+				.route("rewrite_request_obj", r -> r.host("*.rewriterequestobj.org")
+					.filters(f -> f.addRequestHeader("X-TestHeader", "rewrite_request")
+							.modifyRequestBody(String.class, Hello.class,
+									(exchange, s) -> {
+                                        return new Hello(s.toUpperCase());
+                                    })
+					).uri(uri)
+				)
+                .route("rewrite_request_upper", r -> r.host("*.rewriterequestupper.org")
+					.filters(f -> f.addRequestHeader("X-TestHeader", "rewrite_request_upper")
+							.modifyRequestBody(String.class, String.class,
+									(exchange, s) -> {
+                                        return s.toUpperCase();
+                                    })
+					).uri(uri)
+				)
+				.route("rewrite_response_upper", r -> r.host("*.rewriteresponseupper.org")
+					.filters(f -> f.addRequestHeader("X-TestHeader", "rewrite_response_upper")
+							.modifyResponseBody(String.class, String.class,
+									(exchange, s) -> {
+                                        return s.toUpperCase();
+                                    })
+					).uri(uri)
+				)
+                .route("rewrite_response_obj", r -> r.host("*.rewriteresponseobj.org")
+					.filters(f -> f.addRequestHeader("X-TestHeader", "rewrite_response_obj")
+							.modifyResponseBody(Map.class, String.class,
+									(exchange, map) -> {
+										Object data = map.get("data");
+                                        return data.toString();
+                                    })
+					).uri(uri)
 				)
 				.route(r -> r.path("/image/webp")
 					.filters(f ->
 							f.addResponseHeader("X-AnotherHeader", "baz"))
-					.uri("http://httpbin.org:80")
+					.uri(uri)
 				)
 				.route(r -> r.order(-1)
 					.host("**.throttle.org").and().path("/get")
@@ -61,7 +108,7 @@ public class GatewaySampleApplication {
 									.setRefillTokens(1)
 									.setRefillPeriod(10)
 									.setRefillUnit(TimeUnit.SECONDS)))
-					.uri("http://httpbin.org:80")
+					.uri(uri)
 				)
 				.build();
 		//@formatter:on
@@ -75,6 +122,24 @@ public class GatewaySampleApplication {
 		return route;
 	}
 
+	static class Hello {
+		String message;
+
+		public Hello() { }
+
+		public Hello(String message) {
+			this.message = message;
+		}
+
+		public String getMessage() {
+			return message;
+		}
+
+		public void setMessage(String message) {
+			this.message = message;
+		}
+	}
+
 	public static void main(String[] args) {
 		SpringApplication.run(GatewaySampleApplication.class, args);
 	}