Updates to properly dispose of DataBuffers and cleanup connections.

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

@@ -21,6 +21,8 @@ import java.util.List;
 
 import io.netty.handler.codec.http.DefaultHttpHeaders;
 import io.netty.handler.codec.http.HttpMethod;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.netty.NettyPipeline;
@@ -59,6 +61,8 @@ import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.s
  */
 public class NettyRoutingFilter implements GlobalFilter, Ordered {
 
+	private static final Log log = LogFactory.getLog(NettyRoutingFilter.class);
+
 	private final HttpClient httpClient;
 
 	private final ObjectProvider<List<HttpHeadersFilter>> headersFiltersProvider;
@@ -126,11 +130,24 @@ public class NettyRoutingFilter implements GlobalFilter, Ordered {
 						String host = request.getHeaders().getFirst(HttpHeaders.HOST);
 						req.header(HttpHeaders.HOST, host);
 					}
+					if (log.isTraceEnabled()) {
+						nettyOutbound
+								.withConnection(connection -> log.trace("outbound route: "
+										+ connection.channel().id().asShortText()
+										+ ", inbound: " + exchange.getLogPrefix()));
+					}
 					return nettyOutbound.options(NettyPipeline.SendOptions::flushOnEach)
 							.send(request.getBody()
 									.map(dataBuffer -> ((NettyDataBuffer) dataBuffer)
 											.getNativeBuffer()));
 				}).responseConnection((res, connection) -> {
+
+					// Defer committing the response until all route filters have run
+					// Put client response as ServerWebExchange attribute and write
+					// response later NettyWriteResponseFilter
+					exchange.getAttributes().put(CLIENT_RESPONSE_ATTR, res);
+					exchange.getAttributes().put(CLIENT_RESPONSE_CONN_ATTR, connection);
+
 					ServerHttpResponse response = exchange.getResponse();
 					// put headers and status so filters can modify the response
 					HttpHeaders headers = new HttpHeaders();
@@ -154,6 +171,7 @@ public class NettyRoutingFilter implements GlobalFilter, Ordered {
 								.setStatusCodeValue(res.status().code());
 					}
 					else {
+						// TODO: log warning here, not throw error?
 						throw new IllegalStateException(
 								"Unable to set status code on response: "
 										+ res.status().code() + ", "
@@ -181,12 +199,6 @@ public class NettyRoutingFilter implements GlobalFilter, Ordered {
 
 					response.getHeaders().putAll(filteredResponseHeaders);
 
-					// Defer committing the response until all route filters have run
-					// Put client response as ServerWebExchange attribute and write
-					// response later NettyWriteResponseFilter
-					exchange.getAttributes().put(CLIENT_RESPONSE_ATTR, res);
-					exchange.getAttributes().put(CLIENT_RESPONSE_CONN_ATTR, connection);
-
 					return Mono.just(res);
 				});
 

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

@@ -61,34 +61,54 @@ public class NettyWriteResponseFilter implements GlobalFilter, Ordered {
 	public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
 		// NOTICE: nothing in "pre" filter stage as CLIENT_RESPONSE_CONN_ATTR is not added
 		// until the NettyRoutingFilter is run
-		return chain.filter(exchange).then(Mono.defer(() -> {
-			Connection connection = exchange.getAttribute(CLIENT_RESPONSE_CONN_ATTR);
+		// @formatter:off
+		return chain.filter(exchange)
+				.doOnError(throwable -> cleanup(exchange))
+				.then(Mono.defer(() -> {
+					Connection connection = exchange.getAttribute(CLIENT_RESPONSE_CONN_ATTR);
 
-			if (connection == null) {
-				return Mono.empty();
-			}
-			log.trace("NettyWriteResponseFilter start");
-			ServerHttpResponse response = exchange.getResponse();
+					if (connection == null) {
+						return Mono.empty();
+					}
+					if (log.isTraceEnabled()) {
+						log.trace("NettyWriteResponseFilter start inbound: "
+								+ connection.channel().id().asShortText() + ", outbound: "
+								+ exchange.getLogPrefix());
+					}
+					ServerHttpResponse response = exchange.getResponse();
 
-			NettyDataBufferFactory factory = (NettyDataBufferFactory) response
-					.bufferFactory();
-			// TODO: what if it's not netty
+					// TODO: what if it's not netty
+					NettyDataBufferFactory factory = (NettyDataBufferFactory) response
+							.bufferFactory();
 
-			final Flux<NettyDataBuffer> body = connection.inbound().receive().retain() // TODO:
-																						// needed?
-					.map(factory::wrap);
+					// TODO: needed?
+					final Flux<NettyDataBuffer> body = connection
+							.inbound()
+							.receive()
+							.retain()
+							.map(factory::wrap);
 
-			MediaType contentType = null;
-			try {
-				contentType = response.getHeaders().getContentType();
-			}
-			catch (Exception e) {
-				log.trace("invalid media type", e);
-			}
-			return (isStreamingMediaType(contentType)
-					? response.writeAndFlushWith(body.map(Flux::just))
-					: response.writeWith(body));
-		}));
+					MediaType contentType = null;
+					try {
+						contentType = response.getHeaders().getContentType();
+					}
+					catch (Exception e) {
+						if (log.isTraceEnabled()) {
+							log.trace("invalid media type", e);
+						}
+					}
+					return (isStreamingMediaType(contentType)
+							? response.writeAndFlushWith(body.map(Flux::just))
+							: response.writeWith(body));
+				})).doOnCancel(() -> cleanup(exchange));
+		// @formatter:on
+	}
+
+	private void cleanup(ServerWebExchange exchange) {
+		Connection connection = exchange.getAttribute(CLIENT_RESPONSE_CONN_ATTR);
+		if (connection != null) {
+			connection.dispose();
+		}
 	}
 
 	// TODO: use framework if possible

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

@@ -14,9 +14,8 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.gateway.support;
+package org.springframework.cloud.gateway.filter.factory.rewrite;
 
-import java.util.function.Function;
 import java.util.function.Supplier;
 
 import org.reactivestreams.Publisher;
@@ -28,29 +27,21 @@ 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
+ * Implementation of {@link ClientHttpRequest} that saves body as a field.
  */
-public class CachedBodyOutputMessage implements ReactiveHttpOutputMessage {
+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 Flux<DataBuffer> body = Flux.error(new IllegalStateException(
+			"The body is not set. " + "Did handling complete with success?"));
 
-	private Function<Flux<DataBuffer>, Mono<Void>> writeHandler = initDefaultWriteHandler();
-
-	public CachedBodyOutputMessage(ServerWebExchange exchange, HttpHeaders httpHeaders) {
+	CachedBodyOutputMessage(ServerWebExchange exchange, HttpHeaders httpHeaders) {
 		this.bufferFactory = exchange.getResponse().bufferFactory();
 		this.httpHeaders = httpHeaders;
 	}
@@ -70,45 +61,22 @@ public class CachedBodyOutputMessage implements ReactiveHttpOutputMessage {
 		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.
+	 * Return the request body, or an error stream if the body was never set or when.
 	 * @return body as {@link Flux}
 	 */
 	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)));
+		this.body = Flux.from(body);
+		return Mono.empty();
 	}
 
 	@Override

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

@@ -1,112 +0,0 @@
-/*
- * Copyright 2013-2019 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.filter.factory.rewrite;
-
-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.HttpStatus;
-import org.springframework.http.ResponseCookie;
-import org.springframework.http.server.reactive.ServerHttpResponse;
-import org.springframework.util.MultiValueMap;
-
-/**
- * This class is BETA and may be subject to change in a future release. 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

@@ -25,8 +25,6 @@ import reactor.core.publisher.Mono;
 import org.springframework.cloud.gateway.filter.GatewayFilter;
 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.DefaultServerRequest;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.codec.HttpMessageReader;
@@ -60,7 +58,7 @@ public class ModifyRequestBodyGatewayFilterFactory extends
 	public GatewayFilter apply(Config config) {
 		return (exchange, chain) -> {
 			Class inClass = config.getInClass();
-			ServerRequest serverRequest = new DefaultServerRequest(exchange,
+			ServerRequest serverRequest = ServerRequest.create(exchange,
 					this.messageReaders);
 
 			// TODO: flux or mono

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

@@ -27,20 +27,14 @@ 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.io.buffer.DataBuffer;
 import org.springframework.http.HttpHeaders;
-import org.springframework.http.HttpStatus;
-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.reactive.function.client.ClientResponse;
 import org.springframework.web.server.ServerWebExchange;
 
 import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.ORIGINAL_RESPONSE_CONTENT_TYPE_ATTR;
@@ -51,11 +45,13 @@ import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.O
 public class ModifyResponseBodyGatewayFilterFactory extends
 		AbstractGatewayFilterFactory<ModifyResponseBodyGatewayFilterFactory.Config> {
 
-	private final ServerCodecConfigurer codecConfigurer;
-
-	public ModifyResponseBodyGatewayFilterFactory(ServerCodecConfigurer codecConfigurer) {
+	public ModifyResponseBodyGatewayFilterFactory() {
 		super(Config.class);
-		this.codecConfigurer = codecConfigurer;
+	}
+
+	@Deprecated
+	public ModifyResponseBodyGatewayFilterFactory(ServerCodecConfigurer codecConfigurer) {
+		this();
 	}
 
 	@Override
@@ -171,10 +167,11 @@ public class ModifyResponseBodyGatewayFilterFactory extends
 					// types like "Content-Type: image"
 					httpHeaders.add(HttpHeaders.CONTENT_TYPE,
 							originalResponseContentType);
-					ResponseAdapter responseAdapter = new ResponseAdapter(body,
-							httpHeaders);
-					DefaultClientResponse clientResponse = new DefaultClientResponse(
-							responseAdapter, ExchangeStrategies.withDefaults());
+
+					ClientResponse clientResponse = ClientResponse
+							.create(exchange.getResponse().getStatusCode())
+							.headers(headers -> headers.putAll(httpHeaders))
+							.body(Flux.from(body)).build();
 
 					// TODO: flux or mono
 					Mono modifiedBody = clientResponse.bodyToMono(inClass)
@@ -193,7 +190,7 @@ public class ModifyResponseBodyGatewayFilterFactory extends
 									messageBody = messageBody.doOnNext(data -> headers
 											.setContentLength(data.readableByteCount()));
 								}
-								// TODO: use isStreamingMediaType?
+								// TODO: fail if isStreamingMediaType?
 								return getDelegate().writeWith(messageBody);
 							}));
 				}
@@ -215,48 +212,4 @@ public class ModifyResponseBodyGatewayFilterFactory extends
 
 	}
 
-	public class ResponseAdapter implements ClientHttpResponse {
-
-		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;
-		}
-
-		@Override
-		public HttpStatus getStatusCode() {
-			return null;
-		}
-
-		@Override
-		public int getRawStatusCode() {
-			return 0;
-		}
-
-		@Override
-		public MultiValueMap<String, ResponseCookie> getCookies() {
-			return null;
-		}
-
-	}
-
 }

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

@@ -38,7 +38,9 @@ import org.springframework.web.server.ServerWebExchange;
 import static org.springframework.cloud.gateway.filter.AdaptCachedBodyGlobalFilter.CACHED_REQUEST_BODY_KEY;
 
 /**
- * This predicate is BETA and may be subject to change in a future release.
+ * Predicate that reads the body and applies a user provided predicate to run on the body.
+ * The body is cached in memory so that possible subsequent calls to the predicate do not
+ * need to deserialize again.
  */
 public class ReadBodyPredicateFactory
 		extends AbstractRoutePredicateFactory<ReadBodyPredicateFactory.Config> {
@@ -87,19 +89,15 @@ public class ReadBodyPredicateFactory
 				// Join all the DataBuffers so we have a single DataBuffer for the body
 				return DataBufferUtils.join(exchange.getRequest().getBody())
 						.flatMap(dataBuffer -> {
-							// Update the retain counts so we can read the body twice,
-							// once to parse into an object
-							// that we can test the predicate against and a second time
-							// when the HTTP client sends
-							// the request downstream
-							// Note: if we end up reading the body twice we will run into
-							// a problem, but as of right
-							// now there is no good use case for doing this
-							DataBufferUtils.retain(dataBuffer);
-							// Make a slice for each read so each read has its own
-							// read/write indexes
-							Flux<DataBuffer> cachedFlux = Flux.defer(() -> Flux.just(
-									dataBuffer.slice(0, dataBuffer.readableByteCount())));
+							byte[] bytes = new byte[dataBuffer.readableByteCount()];
+							dataBuffer.read(bytes);
+							DataBufferUtils.release(dataBuffer);
+							Flux<DataBuffer> cachedFlux = Flux.defer(() -> {
+								DataBuffer buffer = exchange.getResponse().bufferFactory()
+										.wrap(bytes);
+								DataBufferUtils.retain(buffer);
+								return Mono.just(buffer);
+							});
 
 							ServerHttpRequest mutatedRequest = new ServerHttpRequestDecorator(
 									exchange.getRequest()) {

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/DefaultClientResponse.java

@@ -1,244 +0,0 @@
-/*
- * Copyright 2013-2019 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.support;
-
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.OptionalLong;
-
-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;
-import org.springframework.web.reactive.function.client.ClientResponse;
-import org.springframework.web.reactive.function.client.ExchangeStrategies;
-
-/**
- * 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 int rawStatusCode() {
-		return this.response.getRawStatusCode();
-	}
-
-	@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));
-	}
-
-	@SuppressWarnings("serial")
-	private static class ReadCancellationException extends RuntimeException {
-
-	}
-
-	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());
-		}
-
-	}
-
-}

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/DefaultServerRequest.java

@@ -1,309 +0,0 @@
-/*
- * Copyright 2013-2019 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.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 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.HandlerStrategies;
-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 Optional<InetSocketAddress> remoteAddress() {
-		return Optional.of(request().getRemoteAddress());
-	}
-
-	@Override
-	public List<HttpMessageReader<?>> messageReaders() {
-		return this.messageReaders;
-	}
-
-	@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();
-	}
-
-	public 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();
-		}
-
-	}
-
-}

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/DefaultServerResponse.java

@@ -1,120 +0,0 @@
-/*
- * Copyright 2013-2019 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.support;
-
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-
-import reactor.core.publisher.Mono;
-
-import org.springframework.http.HttpHeaders;
-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.HandlerStrategies;
-import org.springframework.web.reactive.function.server.ServerResponse;
-import org.springframework.web.reactive.result.view.ViewResolver;
-import org.springframework.web.server.ServerWebExchange;
-
-public class DefaultServerResponse<T> implements ServerResponse {
-
-	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();
-		}
-
-	}
-
-}