Updates rewrite filters to use framework web function classes. (#329)

This avoids any blocking operations. fixes gh-316
2018-06-15 20:35:54 -04:00
parent 5d1f111750
commit dacd641335
15 changed files with 1140 additions and 249 deletions
--- 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
@@ -29,24 +29,24 @@ public class AdaptCachedBodyGlobalFilter implements GlobalFilter, Ordered {
 	public static final String CACHED_REQUEST_BODY_KEY = "cachedRequestBody";

 	@Override
-    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
+	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());
-        }
+		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);
-    }
+		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
-    }
+	@Override
+	public int getOrder() {
+		return Ordered.HIGHEST_PRECEDENCE + 1000;
+	}
 }
--- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/NettyRoutingFilter.java
+++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/NettyRoutingFilter.java
@@ -111,9 +111,11 @@ public class NettyRoutingFilter implements GlobalFilter, Ordered {
 			ServerHttpResponse response = exchange.getResponse();
 			// put headers and status so filters can modify the response
 			HttpHeaders headers = new HttpHeaders();
-			
+
 			res.responseHeaders().forEach(entry -> headers.add(entry.getKey(), entry.getValue()));

+			exchange.getAttributes().put("original_response_content_type", headers.getContentType());
+
 			HttpHeaders filteredResponseHeaders = HttpHeadersFilter.filter(
 					this.headersFilters.getIfAvailable(), headers, exchange, Type.RESPONSE);
 			
--- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/rewrite/ModifyRequestBodyGatewayFilterFactory.java
+++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/rewrite/ModifyRequestBodyGatewayFilterFactory.java
@@ -18,24 +18,22 @@
 package org.springframework.cloud.gateway.filter.factory.rewrite;

 import java.util.Map;
-import java.util.Optional;

-import org.reactivestreams.Publisher;
+import org.springframework.cloud.gateway.support.BodyInserterContext;
+import org.springframework.cloud.gateway.support.CachedBodyOutputMessage;
 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.cloud.gateway.support.DefaultServerRequest;
 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;
+import org.springframework.web.reactive.function.BodyInserter;
+import org.springframework.web.reactive.function.BodyInserters;
+import org.springframework.web.reactive.function.server.ServerRequest;

 /**
 * This filter is BETA and may be subject to change in a future release.
@@ -43,11 +41,13 @@ import static org.springframework.cloud.gateway.filter.factory.rewrite.RewriteUt
 public class ModifyRequestBodyGatewayFilterFactory
 		extends AbstractGatewayFilterFactory<ModifyRequestBodyGatewayFilterFactory.Config> {

-	private final ServerCodecConfigurer codecConfigurer;
-
-	public ModifyRequestBodyGatewayFilterFactory(ServerCodecConfigurer codecConfigurer) {
+	public ModifyRequestBodyGatewayFilterFactory() {
 		super(Config.class);
-		this.codecConfigurer = codecConfigurer;
+	}
+
+	@Deprecated
+	public ModifyRequestBodyGatewayFilterFactory(ServerCodecConfigurer codecConfigurer) {
+		this();
 	}

 	@Override
@@ -56,57 +56,36 @@ public class ModifyRequestBodyGatewayFilterFactory
 		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);
+			ServerRequest serverRequest = new DefaultServerRequest(exchange);
+			//TODO: flux or mono
+			Mono<?> modifiedBody = serverRequest.bodyToMono(inClass)
+					// .log("modify_request_mono", Level.INFO)
+					.flatMap(o -> config.rewriteFunction.apply(exchange, o));

-			if (reader.isPresent()) {
-				Mono<Object> readMono = reader.get()
-						.readMono(inElementType, exchange.getRequest(), config.getInHints())
-						.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, config.getOutHints());
+			BodyInserter bodyInserter = BodyInserters.fromPublisher(modifiedBody, config.getOutClass());
+			CachedBodyOutputMessage outputMessage = new CachedBodyOutputMessage(exchange, exchange.getRequest().getHeaders());
+			return bodyInserter.insert(outputMessage,  new BodyInserterContext())
+					// .log("modify_request", Level.INFO)
+					.then(Mono.defer(() -> {
 						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
+								// 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 outputMessage.getBody();
 							}
 						};
 						return chain.filter(exchange.mutate().request(decorator).build());
-					}
-					return chain.filter(exchange);
-				});
+					}));

-			}
-			return chain.filter(exchange);
 		};
 	}

@@ -159,7 +138,7 @@ public class ModifyRequestBodyGatewayFilterFactory
 		}

 		public <T, R> Config setRewriteFunction(Class<T> inClass, Class<R> outClass,
-				RewriteFunction<T, R> rewriteFunction) {
+												RewriteFunction<T, R> rewriteFunction) {
 			setInClass(inClass);
 			setOutClass(outClass);
 			setRewriteFunction(rewriteFunction);
--- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/rewrite/ModifyResponseBodyGatewayFilterFactory.java
+++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/rewrite/ModifyResponseBodyGatewayFilterFactory.java
@@ -18,7 +18,6 @@
 package org.springframework.cloud.gateway.filter.factory.rewrite;

 import java.util.Map;
-import java.util.Optional;

 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
@@ -28,21 +27,24 @@ 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.cloud.gateway.support.BodyInserterContext;
+import org.springframework.cloud.gateway.support.CachedBodyOutputMessage;
+import org.springframework.cloud.gateway.support.DefaultClientResponse;
 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.HttpStatus;
 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.ResponseCookie;
+import org.springframework.http.client.reactive.ClientHttpResponse;
 import org.springframework.http.codec.ServerCodecConfigurer;
 import org.springframework.http.server.reactive.ServerHttpResponseDecorator;
+import org.springframework.util.MultiValueMap;
+import org.springframework.web.reactive.function.BodyInserter;
+import org.springframework.web.reactive.function.BodyInserters;
+import org.springframework.web.reactive.function.client.ExchangeStrategies;
 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;
-
 /**
 * This filter is BETA and may be subject to change in a future release.
 */
@@ -71,34 +73,40 @@ public class ModifyResponseBodyGatewayFilterFactory
 		@Override
 		@SuppressWarnings("unchecked")
 		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);
+					Class inClass = config.getInClass();
+					Class outClass = config.getOutClass();

-					if (reader.isPresent() && writer.isPresent()) {
+					MediaType originalResponseContentType = exchange.getAttribute("original_response_content_type");
+					HttpHeaders httpHeaders = new HttpHeaders();
+					httpHeaders.setContentType(originalResponseContentType);
+					ResponseAdapter responseAdapter = new ResponseAdapter(body, httpHeaders);
+					DefaultClientResponse clientResponse = new DefaultClientResponse(responseAdapter, ExchangeStrategies.withDefaults());

-						ResponseAdapter responseAdapter = new ResponseAdapter(body, getDelegate().getHeaders());
+					//TODO: flux or mono
+					Mono modifiedBody = clientResponse.bodyToMono(inClass)
+							.flatMap(originalBody -> config.rewriteFunction.apply(exchange, originalBody));

-						Flux<?> modified = reader.get().read(inElementType, responseAdapter, config.getInHints())
-								.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(),
-										config.getOutHints())
-						);
-
-					}
-					// TODO: error? log?
-
-					return getDelegate().writeWith(body);
+					BodyInserter bodyInserter = BodyInserters.fromPublisher(modifiedBody, outClass);
+					CachedBodyOutputMessage outputMessage = new CachedBodyOutputMessage(exchange, exchange.getResponse().getHeaders());
+					return bodyInserter.insert(outputMessage, new BodyInserterContext())
+							.then(Mono.defer(() -> {
+								long contentLength1 = getDelegate().getHeaders().getContentLength();
+								Flux<DataBuffer> messageBody = outputMessage.getBody();
+								//TODO: if (inputStream instanceof Mono) {
+									HttpHeaders headers = getDelegate().getHeaders();
+									if (/*headers.getContentLength() < 0 &&*/ !headers.containsKey(HttpHeaders.TRANSFER_ENCODING)) {
+										messageBody = messageBody.doOnNext(data -> headers.setContentLength(data.readableByteCount()));
+									}
+								// }
+								//TODO: use isStreamingMediaType?
+								return getDelegate().writeWith(messageBody);
+							}));
 				}

 				@Override
@@ -118,7 +126,7 @@ public class ModifyResponseBodyGatewayFilterFactory

 	}

-	public class ResponseAdapter implements ReactiveHttpInputMessage {
+	public class ResponseAdapter implements ClientHttpResponse {

 		private final Flux<DataBuffer> flux;
 		private final HttpHeaders headers;
@@ -141,6 +149,21 @@ public class ModifyResponseBodyGatewayFilterFactory
 		public HttpHeaders getHeaders() {
 			return headers;
 		}
+
+		@Override
+		public HttpStatus getStatusCode() {
+			return null;
+		}
+
+		@Override
+		public int getRawStatusCode() {
+			return 0;
+		}
+
+		@Override
+		public MultiValueMap<String, ResponseCookie> getCookies() {
+			return null;
+		}
 	}

 	public static class Config {
@@ -148,6 +171,7 @@ public class ModifyResponseBodyGatewayFilterFactory
 		private Class outClass;
 		private Map<String, Object> inHints;
 		private Map<String, Object> outHints;
+		private String newContentType;

 		private RewriteFunction rewriteFunction;

@@ -187,6 +211,15 @@ public class ModifyResponseBodyGatewayFilterFactory
 			return this;
 		}

+		public String getNewContentType() {
+			return newContentType;
+		}
+
+		public Config setNewContentType(String newContentType) {
+			this.newContentType = newContentType;
+			return this;
+		}
+
 		public RewriteFunction getRewriteFunction() {
 			return rewriteFunction;
 		}
--- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/rewrite/RewriteFunction.java
+++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/rewrite/RewriteFunction.java
@@ -17,14 +17,15 @@

 package org.springframework.cloud.gateway.filter.factory.rewrite;

-import org.springframework.web.server.ServerWebExchange;
-
 import java.util.function.BiFunction;

+import org.reactivestreams.Publisher;
+import org.springframework.web.server.ServerWebExchange;
+
 /**
 * This interface is BETA and may be subject to change in a future release.
 * @param <T>
 * @param <R>
 */
-public interface RewriteFunction<T, R> extends BiFunction<ServerWebExchange, T, R> {
+public interface RewriteFunction<T, R> extends BiFunction<ServerWebExchange, T, Publisher<R>> {
 }
--- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/rewrite/RewriteUtils.java
+++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/rewrite/RewriteUtils.java
@@ -1,70 +0,0 @@
-/*
- * 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;
-
-/**
- * This class is BETA and may be subject to change in a future release.
- */
-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();
-	}
-}
--- 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,94 +18,104 @@
 package org.springframework.cloud.gateway.handler.predicate;

 import java.util.Map;
-import java.util.Optional;
 import java.util.function.Predicate;

-import org.reactivestreams.Publisher;
+import org.springframework.cloud.gateway.support.BodyInserterContext;
 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.cloud.gateway.support.CachedBodyOutputMessage;
+import org.springframework.cloud.gateway.handler.AsyncPredicate;
+import org.springframework.cloud.gateway.support.DefaultServerRequest;
 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.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;

 /**
 * This predicate is BETA and may be subject to change in a future release.
 */
-public class ReadBodyPredicateFactory extends AbstractRoutePredicateFactory<ReadBodyPredicateFactory.Config> {
+public class ReadBodyPredicateFactory
+		extends AbstractRoutePredicateFactory<ReadBodyPredicateFactory.Config> {

+	private static final String TEST_ATTRIBUTE = "read_body_predicate_test_attribute";
 	private final ServerCodecConfigurer codecConfigurer;

-    public ReadBodyPredicateFactory(ServerCodecConfigurer codecConfigurer) {
-        super(Config.class);
+	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(), config.getHints())
-                        .cast(Object.class);
-                answer = process(readMono, peek -> {
-                    Optional<HttpMessageWriter<?>> writer = getHttpMessageWriter(codecConfigurer, elementType, mediaType);
+	@Override
+	@SuppressWarnings("unchecked")
+	public AsyncPredicate<ServerWebExchange> applyAsync(Config config) {
+		return exchange -> {
+			Class inClass = config.getInClass();

-                    if (writer.isPresent()) {
-                        Publisher publisher = Mono.just(peek);
-                        HttpMessageWriterResponse fakeResponse = new HttpMessageWriterResponse(exchange.getResponse().bufferFactory());
-                        writer.get().write(publisher, elementType, mediaType, fakeResponse, config.getHints());
-                        exchange.getAttributes().put(CACHED_REQUEST_BODY_KEY, fakeResponse.getBody());
-                    }
-                    return config.getPredicate().test(peek);
-                });
+			ServerRequest serverRequest = new DefaultServerRequest(exchange);
+			// TODO: flux or mono
+			Mono<?> modifiedBody = serverRequest.bodyToMono(inClass)
+					// .log("modify_request_mono", Level.INFO)
+					.flatMap(body -> {
+						// TODO: migrate to async
+						boolean test = config.predicate.test(body);
+						exchange.getAttributes().put(TEST_ATTRIBUTE, test);
+						return Mono.just(body);
+					});

-            }
-            return answer;
-        };
-    }
+			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,
+								outputMessage.getBody());
+						return Mono.just(test);
+					}));
+		};
+	}

-    public static class Config {
-        private Class inClass;
-        private Predicate predicate;
-        private Map<String, Object> hints;
+	@Override
+	@SuppressWarnings("unchecked")
+	public Predicate<ServerWebExchange> apply(Config config) {
+		throw new UnsupportedOperationException(
+				"ReadBodyPredicateFactory is only async.");
+	}

-        public Class getInClass() {
-            return inClass;
-        }
+	public static class Config {
+		private Class inClass;
+		private Predicate predicate;
+		private Map<String, Object> hints;

-        public Config setInClass(Class inClass) {
-            this.inClass = inClass;
-            return this;
-        }
+		public Class getInClass() {
+			return inClass;
+		}

-        public Predicate getPredicate() {
-            return predicate;
-        }
+		public Config setInClass(Class inClass) {
+			this.inClass = inClass;
+			return this;
+		}

-        public <T> Config setPredicate(Class<T> inClass, Predicate<T> predicate) {
-            setInClass(inClass);
-            this.predicate = predicate;
-            return this;
-        }
+		public Predicate getPredicate() {
+			return predicate;
+		}

-        public Config setPredicate(Predicate predicate) {
-            this.predicate = predicate;
-            return this;
-        }
+		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;
+		}

 		public Map<String, Object> getHints() {
 			return hints;
--- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/route/builder/GatewayFilterSpec.java
+++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/route/builder/GatewayFilterSpec.java
@@ -183,6 +183,7 @@ public class GatewayFilterSpec extends UriSpec {
 	 * @param <R> the new request body class
 	 * @return a {@link GatewayFilterSpec} that can be used to apply additional filters
 	 */
+	//TODO: setup custom spec
 	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)));
@@ -203,6 +204,23 @@ public class GatewayFilterSpec extends UriSpec {
 				.apply(c -> c.setRewriteFunction(inClass, outClass, rewriteFunction)));
 	}

+	/**
+	 * A filter that can be used to modify the response body
+	 * This filter is BETA and may be subject to change in a future release.
+	 * @param inClass the class to conver the response body to
+	 * @param outClass the class the Gateway will add to the response before it is returned to the client
+	 * @param newContentType the new Content-Type header to be returned
+	 * @param rewriteFunction the {@link RewriteFunction} that transforms the response body
+	 * @param <T> the original response body class
+	 * @param <R> the new response body class
+	 * @return a {@link GatewayFilterSpec} that can be used to apply additional filters
+	 */
+	//TODO: setup custom spec
+	public <T, R> GatewayFilterSpec modifyResponseBody(Class<T> inClass, Class<R> outClass, String newContentType, RewriteFunction<T, R> rewriteFunction) {
+		return filter(getBean(ModifyResponseBodyGatewayFilterFactory.class)
+				.apply(c -> c.setRewriteFunction(inClass, outClass, rewriteFunction).setNewContentType(newContentType)));
+	}
+
 	/**
 	 * A filter that can be used to add a prefix to the path of a request before it is routed by the Gateway.
 	 * @param prefix the prefix to add to the path
--- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/BodyInserterContext.java
+++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/BodyInserterContext.java
@@ -0,0 +1,53 @@
+/*
+ * 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.support;
+
+import org.springframework.http.codec.HttpMessageWriter;
+import org.springframework.http.server.reactive.ServerHttpRequest;
+import org.springframework.web.reactive.function.BodyInserter;
+import org.springframework.web.reactive.function.client.ExchangeStrategies;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+public class BodyInserterContext implements BodyInserter.Context {
+	private final ExchangeStrategies exchangeStrategies;
+
+	public BodyInserterContext() {
+		this.exchangeStrategies = ExchangeStrategies.withDefaults();
+	}
+
+	public BodyInserterContext(ExchangeStrategies exchangeStrategies) {
+		this.exchangeStrategies = exchangeStrategies; //TODO: support custom strategies
+	}
+
+	@Override
+	public List<HttpMessageWriter<?>> messageWriters() {
+		return exchangeStrategies.messageWriters();
+	}
+	@Override
+	public Optional<ServerHttpRequest> serverRequest() {
+		return Optional.empty();
+	}
+	@Override
+	public Map<String, Object> hints() {
+		return Collections.emptyMap(); //TODO: support hints
+	}
+}
--- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/CachedBodyOutputMessage.java
+++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/CachedBodyOutputMessage.java
@@ -0,0 +1,122 @@
+/*
+ * 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.support;
+
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.ReactiveHttpOutputMessage;
+import org.springframework.http.client.reactive.ClientHttpRequest;
+import org.springframework.util.Assert;
+import org.springframework.web.server.ServerWebExchange;
+
+/**
+ * Mock implementation of {@link ClientHttpRequest}.
+ * @author Brian Clozel
+ * @author Rossen Stoyanchev
+ * @since 5.0
+ */
+public class CachedBodyOutputMessage implements ReactiveHttpOutputMessage {
+
+	private final DataBufferFactory bufferFactory;
+	private final HttpHeaders httpHeaders;
+
+	private Flux<DataBuffer> body = Flux.error(
+			new IllegalStateException("The body is not set. " +
+					"Did handling complete with success? Is a custom \"writeHandler\" configured?"));
+
+	private Function<Flux<DataBuffer>, Mono<Void>> writeHandler = initDefaultWriteHandler();
+
+	public CachedBodyOutputMessage(ServerWebExchange exchange, HttpHeaders httpHeaders) {
+		this.bufferFactory = exchange.getResponse().bufferFactory();
+		this.httpHeaders = httpHeaders;
+	}
+
+	@Override
+	public void beforeCommit(Supplier<? extends Mono<Void>> action) {
+
+	}
+
+	@Override
+	public boolean isCommitted() {
+		return false;
+	}
+
+	@Override
+	public HttpHeaders getHeaders() {
+		return this.httpHeaders;
+	}
+
+	private Function<Flux<DataBuffer>, Mono<Void>> initDefaultWriteHandler() {
+		return body -> {
+			this.body = body.cache();
+			return this.body.then();
+		};
+	}
+
+	@Override
+	public DataBufferFactory bufferFactory() {
+		return this.bufferFactory;
+	}
+
+	/**
+	 * Return the request body, or an error stream if the body was never set
+	 * or when {@link #setWriteHandler} is configured.
+	 */
+	public Flux<DataBuffer> getBody() {
+		return this.body;
+	}
+
+	/**
+	 * Configure a custom handler for writing the request body.
+	 *
+	 * <p>The default write handler consumes and caches the request body so it
+	 * may be accessed subsequently, e.g. in test assertions. Use this property
+	 * when the request body is an infinite stream.
+	 *
+	 * @param writeHandler the write handler to use returning {@code Mono<Void>}
+	 * when the body has been "written" (i.e. consumed).
+	 */
+	public void setWriteHandler(Function<Flux<DataBuffer>, Mono<Void>> writeHandler) {
+		Assert.notNull(writeHandler, "'writeHandler' is required");
+		this.writeHandler = writeHandler;
+	}
+
+	@Override
+	public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
+		return Mono.defer(() -> this.writeHandler.apply(Flux.from(body)));
+	}
+
+	@Override
+	public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> body) {
+		return writeWith(Flux.from(body).flatMap(p -> p));
+	}
+
+	@Override
+	public Mono<Void> setComplete() {
+		return writeWith(Flux.empty());
+	}
+
+}
--- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/DefaultClientResponse.java
+++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/DefaultClientResponse.java
@@ -0,0 +1,241 @@
+/*
+ * Copyright 2002-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.support;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.OptionalLong;
+
+import org.springframework.web.reactive.function.client.ClientResponse;
+import org.springframework.web.reactive.function.client.ExchangeStrategies;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseCookie;
+import org.springframework.http.ResponseEntity;
+import org.springframework.http.client.reactive.ClientHttpResponse;
+import org.springframework.http.codec.HttpMessageReader;
+import org.springframework.http.server.reactive.ServerHttpResponse;
+import org.springframework.util.MultiValueMap;
+import org.springframework.web.reactive.function.BodyExtractor;
+import org.springframework.web.reactive.function.BodyExtractors;
+
+/**
+ * Default implementation of {@link ClientResponse}.
+ *
+ * @author Arjen Poutsma
+ * @author Brian Clozel
+ * @since 5.0
+ */
+public class DefaultClientResponse implements ClientResponse {
+
+	private final ClientHttpResponse response;
+
+	private final Headers headers;
+
+	private final ExchangeStrategies strategies;
+
+
+	public DefaultClientResponse(ClientHttpResponse response, ExchangeStrategies strategies) {
+		this.response = response;
+		this.strategies = strategies;
+		this.headers = new DefaultHeaders();
+	}
+
+
+	@Override
+	public ExchangeStrategies strategies() {
+		return this.strategies;
+	}
+
+	@Override
+	public HttpStatus statusCode() {
+		return this.response.getStatusCode();
+	}
+
+	@Override
+	public Headers headers() {
+		return this.headers;
+	}
+
+	@Override
+	public MultiValueMap<String, ResponseCookie> cookies() {
+		return this.response.getCookies();
+	}
+
+	@Override
+	public <T> T body(BodyExtractor<T, ? super ClientHttpResponse> extractor) {
+		return extractor.extract(this.response, new BodyExtractor.Context() {
+			@Override
+			public List<HttpMessageReader<?>> messageReaders() {
+				return strategies.messageReaders();
+			}
+			@Override
+			public Optional<ServerHttpResponse> serverResponse() {
+				return Optional.empty();
+			}
+			@Override
+			public Map<String, Object> hints() {
+				return Collections.emptyMap();
+			}
+		});
+	}
+
+	@Override
+	public <T> Mono<T> bodyToMono(Class<? extends T> elementClass) {
+		if (Void.class.isAssignableFrom(elementClass)) {
+			return consumeAndCancel();
+		}
+		else {
+			return body(BodyExtractors.toMono(elementClass));
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	private <T> Mono<T> consumeAndCancel() {
+		return (Mono<T>) this.response.getBody()
+				.map(buffer -> {
+					DataBufferUtils.release(buffer);
+					throw new ReadCancellationException();
+				})
+				.onErrorResume(ReadCancellationException.class, ex -> Mono.empty())
+				.then();
+	}
+
+	@Override
+	public <T> Mono<T> bodyToMono(ParameterizedTypeReference<T> typeReference) {
+		if (Void.class.isAssignableFrom(typeReference.getType().getClass())) {
+			return consumeAndCancel();
+		}
+		else {
+			return body(BodyExtractors.toMono(typeReference));
+		}
+	}
+
+	@Override
+	public <T> Flux<T> bodyToFlux(Class<? extends T> elementClass) {
+		if (Void.class.isAssignableFrom(elementClass)) {
+			return Flux.from(consumeAndCancel());
+		}
+		else {
+			return body(BodyExtractors.toFlux(elementClass));
+		}
+	}
+
+	@Override
+	public <T> Flux<T> bodyToFlux(ParameterizedTypeReference<T> typeReference) {
+		if (Void.class.isAssignableFrom(typeReference.getType().getClass())) {
+			return Flux.from(consumeAndCancel());
+		}
+		else {
+			return body(BodyExtractors.toFlux(typeReference));
+		}
+	}
+
+	@Override
+	public <T> Mono<ResponseEntity<T>> toEntity(Class<T> bodyType) {
+		if (Void.class.isAssignableFrom(bodyType)) {
+			return toEntityInternal(consumeAndCancel());
+		}
+		else {
+			return toEntityInternal(bodyToMono(bodyType));
+		}
+	}
+
+	@Override
+	public <T> Mono<ResponseEntity<T>> toEntity(ParameterizedTypeReference<T> typeReference) {
+		if (Void.class.isAssignableFrom(typeReference.getType().getClass())) {
+			return toEntityInternal(consumeAndCancel());
+		}
+		else {
+			return toEntityInternal(bodyToMono(typeReference));
+		}
+	}
+
+	private <T> Mono<ResponseEntity<T>> toEntityInternal(Mono<T> bodyMono) {
+		HttpHeaders headers = headers().asHttpHeaders();
+		HttpStatus statusCode = statusCode();
+		return bodyMono
+				.map(body -> new ResponseEntity<>(body, headers, statusCode))
+				.switchIfEmpty(Mono.defer(
+						() -> Mono.just(new ResponseEntity<>(headers, statusCode))));
+	}
+
+	@Override
+	public <T> Mono<ResponseEntity<List<T>>> toEntityList(Class<T> responseType) {
+		return toEntityListInternal(bodyToFlux(responseType));
+	}
+
+	@Override
+	public <T> Mono<ResponseEntity<List<T>>> toEntityList(ParameterizedTypeReference<T> typeReference) {
+		return toEntityListInternal(bodyToFlux(typeReference));
+	}
+
+	private <T> Mono<ResponseEntity<List<T>>> toEntityListInternal(Flux<T> bodyFlux) {
+		HttpHeaders headers = headers().asHttpHeaders();
+		HttpStatus statusCode = statusCode();
+		return bodyFlux
+				.collectList()
+				.map(body -> new ResponseEntity<>(body, headers, statusCode));
+	}
+
+
+	private class DefaultHeaders implements Headers {
+
+		private HttpHeaders delegate() {
+			return response.getHeaders();
+		}
+
+		@Override
+		public OptionalLong contentLength() {
+			return toOptionalLong(delegate().getContentLength());
+		}
+
+		@Override
+		public Optional<MediaType> contentType() {
+			return Optional.ofNullable(delegate().getContentType());
+		}
+
+		@Override
+		public List<String> header(String headerName) {
+			List<String> headerValues = delegate().get(headerName);
+			return (headerValues != null ? headerValues : Collections.emptyList());
+		}
+
+		@Override
+		public HttpHeaders asHttpHeaders() {
+			return HttpHeaders.readOnlyHttpHeaders(delegate());
+		}
+
+		private OptionalLong toOptionalLong(long value) {
+			return (value != -1 ? OptionalLong.of(value) : OptionalLong.empty());
+		}
+	}
+
+
+	@SuppressWarnings("serial")
+	private static class ReadCancellationException extends RuntimeException {
+	}
+
+}
--- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/DefaultServerRequest.java
+++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/DefaultServerRequest.java
@@ -0,0 +1,297 @@
+/*
+ * 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.support;
+
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.nio.charset.Charset;
+import java.security.Principal;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.function.Function;
+
+import org.springframework.web.reactive.function.server.HandlerStrategies;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.http.HttpCookie;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpRange;
+import org.springframework.http.HttpRequest;
+import org.springframework.http.MediaType;
+import org.springframework.http.codec.HttpMessageReader;
+import org.springframework.http.codec.multipart.Part;
+import org.springframework.http.server.PathContainer;
+import org.springframework.http.server.reactive.ServerHttpRequest;
+import org.springframework.http.server.reactive.ServerHttpResponse;
+import org.springframework.util.MultiValueMap;
+import org.springframework.web.reactive.function.BodyExtractor;
+import org.springframework.web.reactive.function.BodyExtractors;
+import org.springframework.web.reactive.function.UnsupportedMediaTypeException;
+import org.springframework.web.reactive.function.server.RouterFunctions;
+import org.springframework.web.reactive.function.server.ServerRequest;
+import org.springframework.web.server.ServerWebExchange;
+import org.springframework.web.server.UnsupportedMediaTypeStatusException;
+import org.springframework.web.server.WebSession;
+import org.springframework.web.util.UriBuilder;
+import org.springframework.web.util.UriComponentsBuilder;
+
+/**
+ * {@code ServerRequest} implementation based on a {@link ServerWebExchange}.
+ *
+ * @author Arjen Poutsma
+ * @since 5.0
+ */
+public class DefaultServerRequest implements ServerRequest {
+
+	private static final Function<UnsupportedMediaTypeException, UnsupportedMediaTypeStatusException> ERROR_MAPPER =
+			ex -> (ex.getContentType() != null ?
+					new UnsupportedMediaTypeStatusException(ex.getContentType(), ex.getSupportedMediaTypes()) :
+					new UnsupportedMediaTypeStatusException(ex.getMessage()));
+
+
+	private final ServerWebExchange exchange;
+
+	private final Headers headers;
+
+	private final List<HttpMessageReader<?>> messageReaders;
+
+	public DefaultServerRequest(ServerWebExchange exchange) {
+		this(exchange, HandlerStrategies.withDefaults().messageReaders());
+	}
+
+	public DefaultServerRequest(ServerWebExchange exchange, List<HttpMessageReader<?>> messageReaders) {
+		this.exchange = exchange;
+		this.messageReaders = Collections.unmodifiableList(new ArrayList<>(messageReaders));
+		this.headers = new DefaultHeaders();
+	}
+
+
+	@Override
+	public String methodName() {
+		return request().getMethodValue();
+	}
+
+	@Override
+	public URI uri() {
+		return request().getURI();
+	}
+
+	@Override
+	public UriBuilder uriBuilder() {
+		return UriComponentsBuilder.fromHttpRequest(new ServerRequestAdapter());
+	}
+
+	@Override
+	public PathContainer pathContainer() {
+		return request().getPath();
+	}
+
+	@Override
+	public Headers headers() {
+		return this.headers;
+	}
+
+	@Override
+	public MultiValueMap<String, HttpCookie> cookies() {
+		return request().getCookies();
+	}
+
+	@Override
+	public <T> T body(BodyExtractor<T, ? super ServerHttpRequest> extractor) {
+		return body(extractor, Collections.emptyMap());
+	}
+
+	@Override
+	public <T> T body(BodyExtractor<T, ? super ServerHttpRequest> extractor, Map<String, Object> hints) {
+		return extractor.extract(request(),
+				new BodyExtractor.Context() {
+					@Override
+					public List<HttpMessageReader<?>> messageReaders() {
+						return messageReaders;
+					}
+					@Override
+					public Optional<ServerHttpResponse> serverResponse() {
+						return Optional.of(exchange().getResponse());
+					}
+					@Override
+					public Map<String, Object> hints() {
+						return hints;
+					}
+				});
+	}
+
+	@Override
+	public <T> Mono<T> bodyToMono(Class<? extends T> elementClass) {
+		Mono<T> mono = body(BodyExtractors.toMono(elementClass));
+		return mono.onErrorMap(UnsupportedMediaTypeException.class, ERROR_MAPPER);
+	}
+
+	@Override
+	public <T> Mono<T> bodyToMono(ParameterizedTypeReference<T> typeReference) {
+		Mono<T> mono = body(BodyExtractors.toMono(typeReference));
+		return mono.onErrorMap(UnsupportedMediaTypeException.class, ERROR_MAPPER);
+	}
+
+	@Override
+	public <T> Flux<T> bodyToFlux(Class<? extends T> elementClass) {
+		Flux<T> flux = body(BodyExtractors.toFlux(elementClass));
+		return flux.onErrorMap(UnsupportedMediaTypeException.class, ERROR_MAPPER);
+	}
+
+	@Override
+	public <T> Flux<T> bodyToFlux(ParameterizedTypeReference<T> typeReference) {
+		Flux<T> flux = body(BodyExtractors.toFlux(typeReference));
+		return flux.onErrorMap(UnsupportedMediaTypeException.class, ERROR_MAPPER);
+	}
+
+	@Override
+	public Map<String, Object> attributes() {
+		return this.exchange.getAttributes();
+	}
+
+	@Override
+	public MultiValueMap<String, String> queryParams() {
+		return request().getQueryParams();
+	}
+
+	@Override
+	public Map<String, String> pathVariables() {
+		return this.exchange.getAttributeOrDefault(
+				RouterFunctions.URI_TEMPLATE_VARIABLES_ATTRIBUTE, Collections.emptyMap());
+	}
+
+	@Override
+	public Mono<WebSession> session() {
+		return this.exchange.getSession();
+	}
+
+	@Override
+	public Mono<? extends Principal> principal() {
+		return this.exchange.getPrincipal();
+	}
+
+	@Override
+	public Mono<MultiValueMap<String, String>> formData() {
+		return this.exchange.getFormData();
+	}
+
+	@Override
+	public Mono<MultiValueMap<String, Part>> multipartData() {
+		return this.exchange.getMultipartData();
+	}
+
+	private ServerHttpRequest request() {
+		return this.exchange.getRequest();
+	}
+
+	ServerWebExchange exchange() {
+		return this.exchange;
+	}
+
+	@Override
+	public String toString() {
+		return String.format("%s %s", method(), path());
+	}
+
+
+	private class DefaultHeaders implements Headers {
+
+		private HttpHeaders delegate() {
+			return request().getHeaders();
+		}
+
+		@Override
+		public List<MediaType> accept() {
+			return delegate().getAccept();
+		}
+
+		@Override
+		public List<Charset> acceptCharset() {
+			return delegate().getAcceptCharset();
+		}
+
+		@Override
+		public List<Locale.LanguageRange> acceptLanguage() {
+			return delegate().getAcceptLanguage();
+		}
+
+		@Override
+		public OptionalLong contentLength() {
+			long value = delegate().getContentLength();
+			return (value != -1 ? OptionalLong.of(value) : OptionalLong.empty());
+		}
+
+		@Override
+		public Optional<MediaType> contentType() {
+			return Optional.ofNullable(delegate().getContentType());
+		}
+
+		@Override
+		public InetSocketAddress host() {
+			return delegate().getHost();
+		}
+
+		@Override
+		public List<HttpRange> range() {
+			return delegate().getRange();
+		}
+
+		@Override
+		public List<String> header(String headerName) {
+			List<String> headerValues = delegate().get(headerName);
+			return (headerValues != null ? headerValues : Collections.emptyList());
+		}
+
+		@Override
+		public HttpHeaders asHttpHeaders() {
+			return HttpHeaders.readOnlyHttpHeaders(delegate());
+		}
+
+		@Override
+		public String toString() {
+			return delegate().toString();
+		}
+	}
+
+
+	private final class ServerRequestAdapter implements HttpRequest {
+
+		@Override
+		public String getMethodValue() {
+			return methodName();
+		}
+
+		@Override
+		public URI getURI() {
+			return uri();
+		}
+
+		@Override
+		public HttpHeaders getHeaders() {
+			return request().getHeaders();
+		}
+	}
+
+}
--- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/DefaultServerResponse.java
+++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/DefaultServerResponse.java
@@ -0,0 +1,123 @@
+/*
+ * 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.support;
+
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import org.springframework.web.reactive.function.server.HandlerStrategies;
+import org.springframework.web.reactive.result.view.ViewResolver;
+import reactor.core.publisher.Mono;
+
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseCookie;
+import org.springframework.http.codec.HttpMessageWriter;
+import org.springframework.http.server.reactive.ServerHttpRequest;
+import org.springframework.http.server.reactive.ServerHttpResponse;
+import org.springframework.util.Assert;
+import org.springframework.util.MultiValueMap;
+import org.springframework.web.reactive.function.BodyInserter;
+import org.springframework.web.reactive.function.server.ServerResponse;
+import org.springframework.web.server.ServerWebExchange;
+
+public class DefaultServerResponse<T> implements ServerResponse {
+
+
+	private static final Set<HttpMethod> SAFE_METHODS = EnumSet.of(HttpMethod.GET, HttpMethod.HEAD);
+
+	private final ServerWebExchange exchange;
+
+	private final BodyInserter<T, ? super ServerHttpResponse> inserter;
+
+	private final Map<String, Object> hints;
+
+	public DefaultServerResponse(ServerWebExchange exchange,
+			BodyInserter<T, ? super ServerHttpResponse> body, Map<String, Object> hints) {
+		this.exchange = exchange;
+		Assert.notNull(exchange, "ServerWebExchange must not be null");
+		Assert.notNull(body, "BodyInserter must not be null");
+		this.inserter = body;
+		this.hints = hints;
+	}
+
+	private ServerHttpResponse response() {
+		return exchange.getResponse();
+	}
+
+	@Override
+	public final HttpStatus statusCode() {
+		//TODO: non standard status code
+		return HttpStatus.valueOf(response().getStatusCode().value());
+	}
+
+	@Override
+	public final HttpHeaders headers() {
+		return response().getHeaders();
+	}
+
+	@Override
+	public MultiValueMap<String, ResponseCookie> cookies() {
+		return response().getCookies();
+	}
+
+	@Override
+	public final Mono<Void> writeTo(ServerWebExchange exchange, Context context) {
+		return this.inserter.insert(exchange.getResponse(), new BodyInserter.Context() {
+			@Override
+			public List<HttpMessageWriter<?>> messageWriters() {
+				return context.messageWriters();
+			}
+			@Override
+			public Optional<ServerHttpRequest> serverRequest() {
+				return Optional.of(exchange.getRequest());
+			}
+			@Override
+			public Map<String, Object> hints() {
+				return hints;
+			}
+		});
+	}
+
+
+	public static class HandlerStrategiesResponseContext implements ServerResponse.Context {
+
+		private HandlerStrategies strategies = HandlerStrategies.withDefaults();
+
+		public HandlerStrategiesResponseContext() {
+		}
+
+		public HandlerStrategiesResponseContext(HandlerStrategies strategies) {
+			this.strategies = strategies;
+		}
+
+		@Override
+		public List<HttpMessageWriter<?>> messageWriters() {
+			return this.strategies.messageWriters();
+		}
+
+		@Override
+		public List<ViewResolver> viewResolvers() {
+			return this.strategies.viewResolvers();
+		}
+	}
+}
--- a/spring-cloud-gateway-sample/src/main/java/org/springframework/cloud/gateway/sample/GatewaySampleApplication.java
+++ b/spring-cloud-gateway-sample/src/main/java/org/springframework/cloud/gateway/sample/GatewaySampleApplication.java
@@ -20,6 +20,8 @@ package org.springframework.cloud.gateway.sample;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;

+import reactor.core.publisher.Mono;
+
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.SpringBootConfiguration;
@@ -28,6 +30,7 @@ import org.springframework.cloud.gateway.route.RouteLocator;
 import org.springframework.cloud.gateway.route.builder.RouteLocatorBuilder;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Import;
+import org.springframework.http.MediaType;
 import org.springframework.web.reactive.function.BodyInserters;
 import org.springframework.web.reactive.function.server.RequestPredicates;
 import org.springframework.web.reactive.function.server.RouterFunction;
@@ -59,47 +62,49 @@ public class GatewaySampleApplication {
 				)
 				.route("read_body_pred", r -> r.host("*.readbody.org")
 						.and().readBody(String.class,
-										s -> s.trim().equalsIgnoreCase("hello"))
-					.filters(f ->
-							f.prefixPath("/httpbin")
-									.addRequestHeader("X-TestHeader", "read_body_pred")
+										s -> s.trim().equalsIgnoreCase("hi"))
+					.filters(f -> f.prefixPath("/httpbin")
+							.addResponseHeader("X-TestHeader", "read_body_pred")
 					).uri(uri)
 				)
 				.route("rewrite_request_obj", r -> r.host("*.rewriterequestobj.org")
 					.filters(f -> f.prefixPath("/httpbin")
-									.addRequestHeader("X-TestHeader", "rewrite_request")
+							//TODO: add as configuration to modifyRequestBody
+							.setRequestHeader("Content-Type", MediaType.APPLICATION_JSON_VALUE)
+							.addResponseHeader("X-TestHeader", "rewrite_request")
 							.modifyRequestBody(String.class, Hello.class,
 									(exchange, s) -> {
-                                        return new Hello(s.toUpperCase());
-                                    })
+										return Mono.just(new Hello(s.toUpperCase()));
+									})
 					).uri(uri)
 				)
-                .route("rewrite_request_upper", r -> r.host("*.rewriterequestupper.org")
+				.route("rewrite_request_upper", r -> r.host("*.rewriterequestupper.org")
 					.filters(f -> f.prefixPath("/httpbin")
-									.addRequestHeader("X-TestHeader", "rewrite_request_upper")
+							.addResponseHeader("X-TestHeader", "rewrite_request_upper")
 							.modifyRequestBody(String.class, String.class,
 									(exchange, s) -> {
-                                        return s.toUpperCase();
-                                    })
+										return Mono.just(s.toUpperCase());
+									})
 					).uri(uri)
 				)
 				.route("rewrite_response_upper", r -> r.host("*.rewriteresponseupper.org")
 					.filters(f -> f.prefixPath("/httpbin")
-									.addRequestHeader("X-TestHeader", "rewrite_response_upper")
+							.addResponseHeader("X-TestHeader", "rewrite_response_upper")
 							.modifyResponseBody(String.class, String.class,
 									(exchange, s) -> {
-                                        return s.toUpperCase();
-                                    })
+										return Mono.just(s.toUpperCase());
+									})
 					).uri(uri)
 				)
-                .route("rewrite_response_obj", r -> r.host("*.rewriteresponseobj.org")
+				.route("rewrite_response_obj", r -> r.host("*.rewriteresponseobj.org")
 					.filters(f -> f.prefixPath("/httpbin")
-									.addRequestHeader("X-TestHeader", "rewrite_response_obj")
-							.modifyResponseBody(Map.class, String.class,
+							.addResponseHeader("X-TestHeader", "rewrite_response_obj")
+							.modifyResponseBody(Map.class, String.class, MediaType.TEXT_PLAIN_VALUE,
 									(exchange, map) -> {
 										Object data = map.get("data");
-                                        return data.toString();
-                                    })
+										return Mono.just(data.toString());
+									})
+							.setResponseHeader("Content-Type", MediaType.TEXT_PLAIN_VALUE)
 					).uri(uri)
 				)
 				.route(r -> r.path("/image/webp")
--- a/spring-cloud-gateway-sample/src/test/java/org/springframework/cloud/gateway/sample/GatewaySampleApplicationTests.java
+++ b/spring-cloud-gateway-sample/src/test/java/org/springframework/cloud/gateway/sample/GatewaySampleApplicationTests.java
@@ -18,6 +18,7 @@
 package org.springframework.cloud.gateway.sample;

 import java.time.Duration;
+import java.util.Map;

 import com.netflix.loadbalancer.Server;
 import com.netflix.loadbalancer.ServerList;
@@ -41,6 +42,7 @@ import org.springframework.test.context.junit4.SpringRunner;
 import org.springframework.test.web.reactive.server.WebTestClient;
 import org.springframework.util.SocketUtils;

+import static org.assertj.core.api.Assertions.assertThat;
 import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT;

 /**
@@ -85,6 +87,81 @@ public class GatewaySampleApplicationTests {
 				.expectStatus().isOk();
 	}

+	@Test
+	@SuppressWarnings("unchecked")
+	public void readBodyPredicateStringWorks() {
+		webClient.post()
+				.uri("/post")
+				.header("Host", "www.readbody.org")
+				.syncBody("hi")
+				.exchange()
+				.expectStatus().isOk()
+				.expectHeader().valueEquals("X-TestHeader", "read_body_pred")
+				.expectBody(Map.class)
+				.consumeWith(result ->
+						assertThat(result.getResponseBody()).containsEntry("data", "hi"));
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void rewriteRequestBodyStringWorks() {
+		webClient.post()
+				.uri("/post")
+				.header("Host", "www.rewriterequestupper.org")
+				.syncBody("hello")
+				.exchange()
+				.expectStatus().isOk()
+				.expectHeader().valueEquals("X-TestHeader", "rewrite_request_upper")
+				.expectBody(Map.class)
+				.consumeWith(result ->
+						assertThat(result.getResponseBody()).containsEntry("data", "HELLO"));
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void rewriteRequestBodyObjectWorks() {
+		webClient.post()
+				.uri("/post")
+				.header("Host", "www.rewriterequestobj.org")
+				.syncBody("hello")
+				.exchange()
+				.expectStatus().isOk()
+				.expectHeader().valueEquals("X-TestHeader", "rewrite_request")
+				.expectBody(Map.class)
+				.consumeWith(result ->
+						assertThat(result.getResponseBody()).containsEntry("data", "{\"message\":\"HELLO\"}"));
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void rewriteResponseBodyStringWorks() {
+		webClient.post()
+				.uri("/post")
+				.header("Host", "www.rewriteresponseupper.org")
+				.syncBody("hello")
+				.exchange()
+				.expectStatus().isOk()
+				.expectHeader().valueEquals("X-TestHeader", "rewrite_response_upper")
+				.expectBody(Map.class)
+				.consumeWith(result ->
+						assertThat(result.getResponseBody()).containsEntry("DATA", "HELLO"));
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void rewriteResponeBodyObjectWorks() {
+		webClient.post()
+				.uri("/post")
+				.header("Host", "www.rewriteresponseobj.org")
+				.syncBody("hello")
+				.exchange()
+				.expectStatus().isOk()
+				.expectHeader().valueEquals("X-TestHeader", "rewrite_response_obj")
+				.expectBody(String.class)
+				.consumeWith(result ->
+						assertThat(result.getResponseBody()).isEqualTo("hello"));
+	}
+
 	@Test
 	public void complexPredicate() {
 		webClient.get()