diff --git 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 index 4925a21c..5ad44caa 100644 --- 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 @@ -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> 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); }); diff --git a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/NettyWriteResponseFilter.java b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/NettyWriteResponseFilter.java index 7d96a7ab..482723d7 100644 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/NettyWriteResponseFilter.java +++ b/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 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 body = connection.inbound().receive().retain() // TODO: - // needed? - .map(factory::wrap); + // TODO: needed? + final Flux 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 diff --git 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/filter/factory/rewrite/CachedBodyOutputMessage.java similarity index 57% rename from spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/CachedBodyOutputMessage.java rename to spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/rewrite/CachedBodyOutputMessage.java index 37362140..085f8283 100644 --- 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/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 body = Flux - .error(new IllegalStateException("The body is not set. " - + "Did handling complete with success? Is a custom \"writeHandler\" configured?")); + private Flux body = Flux.error(new IllegalStateException( + "The body is not set. " + "Did handling complete with success?")); - private Function, Mono> 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, Mono> 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 getBody() { return this.body; } - /** - * Configure a custom handler for writing the request body. - * - *

- * 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} when the - * body has been "written" (i.e. consumed). - */ - public void setWriteHandler(Function, Mono> writeHandler) { - Assert.notNull(writeHandler, "'writeHandler' is required"); - this.writeHandler = writeHandler; - } - - @Override public Mono writeWith(Publisher body) { - return Mono.defer(() -> this.writeHandler.apply(Flux.from(body))); + this.body = Flux.from(body); + return Mono.empty(); } @Override diff --git a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/rewrite/HttpMessageWriterResponse.java b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/rewrite/HttpMessageWriterResponse.java deleted file mode 100644 index 75119bfd..00000000 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/rewrite/HttpMessageWriterResponse.java +++ /dev/null @@ -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 body; - - public HttpMessageWriterResponse(DataBufferFactory dataBufferFactory) { - this.dataBufferFactory = dataBufferFactory; - } - - @Override - public HttpHeaders getHeaders() { - return this.headers; - } - - @Override - public Mono writeWith(Publisher body) { - this.body = body; - return Mono.empty(); - } - - @Override - public Mono writeAndFlushWith( - Publisher> body) { - // TODO: is this kosher? - return writeWith(Flux.from(body).flatMapSequential(p -> p)); - } - - public Publisher getBody() { - return body; - } - - @Override - public boolean setStatusCode(HttpStatus status) { - return false; - } - - @Override - public HttpStatus getStatusCode() { - return null; - } - - @Override - public MultiValueMap getCookies() { - return null; - } - - @Override - public void addCookie(ResponseCookie cookie) { - - } - - @Override - public DataBufferFactory bufferFactory() { - return this.dataBufferFactory; - } - - @Override - public void beforeCommit(Supplier> action) { - - } - - @Override - public boolean isCommitted() { - return false; - } - - @Override - public Mono setComplete() { - return null; - } - -} diff --git 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 index cd6f5082..ddb21304 100644 --- 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 @@ -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 diff --git 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 index 4a0a33cb..ee95ab5f 100644 --- 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 @@ -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 { - 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 flux; - - private final HttpHeaders headers; - - public ResponseAdapter(Publisher body, - HttpHeaders headers) { - this.headers = headers; - if (body instanceof Flux) { - flux = (Flux) body; - } - else { - flux = ((Mono) body).flux(); - } - } - - @Override - public Flux getBody() { - return flux; - } - - @Override - public HttpHeaders getHeaders() { - return headers; - } - - @Override - public HttpStatus getStatusCode() { - return null; - } - - @Override - public int getRawStatusCode() { - return 0; - } - - @Override - public MultiValueMap getCookies() { - return null; - } - - } - } diff --git a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/handler/predicate/ReadBodyPredicateFactory.java b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/handler/predicate/ReadBodyPredicateFactory.java index 4bf92792..12384b77 100644 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/handler/predicate/ReadBodyPredicateFactory.java +++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/handler/predicate/ReadBodyPredicateFactory.java @@ -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 { @@ -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 cachedFlux = Flux.defer(() -> Flux.just( - dataBuffer.slice(0, dataBuffer.readableByteCount()))); + byte[] bytes = new byte[dataBuffer.readableByteCount()]; + dataBuffer.read(bytes); + DataBufferUtils.release(dataBuffer); + Flux cachedFlux = Flux.defer(() -> { + DataBuffer buffer = exchange.getResponse().bufferFactory() + .wrap(bytes); + DataBufferUtils.retain(buffer); + return Mono.just(buffer); + }); ServerHttpRequest mutatedRequest = new ServerHttpRequestDecorator( exchange.getRequest()) { diff --git 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 deleted file mode 100644 index 0aa9b892..00000000 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/DefaultClientResponse.java +++ /dev/null @@ -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 cookies() { - return this.response.getCookies(); - } - - @Override - public T body(BodyExtractor extractor) { - return extractor.extract(this.response, new BodyExtractor.Context() { - @Override - public List> messageReaders() { - return strategies.messageReaders(); - } - - @Override - public Optional serverResponse() { - return Optional.empty(); - } - - @Override - public Map hints() { - return Collections.emptyMap(); - } - }); - } - - @Override - public Mono bodyToMono(Class elementClass) { - if (Void.class.isAssignableFrom(elementClass)) { - return consumeAndCancel(); - } - else { - return body(BodyExtractors.toMono(elementClass)); - } - } - - @SuppressWarnings("unchecked") - private Mono consumeAndCancel() { - return (Mono) this.response.getBody().map(buffer -> { - DataBufferUtils.release(buffer); - throw new ReadCancellationException(); - }).onErrorResume(ReadCancellationException.class, ex -> Mono.empty()).then(); - } - - @Override - public Mono bodyToMono(ParameterizedTypeReference typeReference) { - if (Void.class.isAssignableFrom(typeReference.getType().getClass())) { - return consumeAndCancel(); - } - else { - return body(BodyExtractors.toMono(typeReference)); - } - } - - @Override - public Flux bodyToFlux(Class elementClass) { - if (Void.class.isAssignableFrom(elementClass)) { - return Flux.from(consumeAndCancel()); - } - else { - return body(BodyExtractors.toFlux(elementClass)); - } - } - - @Override - public Flux bodyToFlux(ParameterizedTypeReference typeReference) { - if (Void.class.isAssignableFrom(typeReference.getType().getClass())) { - return Flux.from(consumeAndCancel()); - } - else { - return body(BodyExtractors.toFlux(typeReference)); - } - } - - @Override - public Mono> toEntity(Class bodyType) { - if (Void.class.isAssignableFrom(bodyType)) { - return toEntityInternal(consumeAndCancel()); - } - else { - return toEntityInternal(bodyToMono(bodyType)); - } - } - - @Override - public Mono> toEntity( - ParameterizedTypeReference typeReference) { - if (Void.class.isAssignableFrom(typeReference.getType().getClass())) { - return toEntityInternal(consumeAndCancel()); - } - else { - return toEntityInternal(bodyToMono(typeReference)); - } - } - - private Mono> toEntityInternal(Mono 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 Mono>> toEntityList(Class responseType) { - return toEntityListInternal(bodyToFlux(responseType)); - } - - @Override - public Mono>> toEntityList( - ParameterizedTypeReference typeReference) { - return toEntityListInternal(bodyToFlux(typeReference)); - } - - private Mono>> toEntityListInternal(Flux 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 contentType() { - return Optional.ofNullable(delegate().getContentType()); - } - - @Override - public List header(String headerName) { - List 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()); - } - - } - -} diff --git 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 deleted file mode 100644 index 96958377..00000000 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/DefaultServerRequest.java +++ /dev/null @@ -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 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> messageReaders; - - public DefaultServerRequest(ServerWebExchange exchange) { - this(exchange, HandlerStrategies.withDefaults().messageReaders()); - } - - public DefaultServerRequest(ServerWebExchange exchange, - List> 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 cookies() { - return request().getCookies(); - } - - @Override - public T body(BodyExtractor extractor) { - return body(extractor, Collections.emptyMap()); - } - - @Override - public Optional remoteAddress() { - return Optional.of(request().getRemoteAddress()); - } - - @Override - public List> messageReaders() { - return this.messageReaders; - } - - @Override - public T body(BodyExtractor extractor, - Map hints) { - return extractor.extract(request(), new BodyExtractor.Context() { - @Override - public List> messageReaders() { - return messageReaders; - } - - @Override - public Optional serverResponse() { - return Optional.of(exchange().getResponse()); - } - - @Override - public Map hints() { - return hints; - } - }); - } - - @Override - public Mono bodyToMono(Class elementClass) { - Mono mono = body(BodyExtractors.toMono(elementClass)); - return mono.onErrorMap(UnsupportedMediaTypeException.class, ERROR_MAPPER); - } - - @Override - public Mono bodyToMono(ParameterizedTypeReference typeReference) { - Mono mono = body(BodyExtractors.toMono(typeReference)); - return mono.onErrorMap(UnsupportedMediaTypeException.class, ERROR_MAPPER); - } - - @Override - public Flux bodyToFlux(Class elementClass) { - Flux flux = body(BodyExtractors.toFlux(elementClass)); - return flux.onErrorMap(UnsupportedMediaTypeException.class, ERROR_MAPPER); - } - - @Override - public Flux bodyToFlux(ParameterizedTypeReference typeReference) { - Flux flux = body(BodyExtractors.toFlux(typeReference)); - return flux.onErrorMap(UnsupportedMediaTypeException.class, ERROR_MAPPER); - } - - @Override - public Map attributes() { - return this.exchange.getAttributes(); - } - - @Override - public MultiValueMap queryParams() { - return request().getQueryParams(); - } - - @Override - public Map pathVariables() { - return this.exchange.getAttributeOrDefault( - RouterFunctions.URI_TEMPLATE_VARIABLES_ATTRIBUTE, Collections.emptyMap()); - } - - @Override - public Mono session() { - return this.exchange.getSession(); - } - - @Override - public Mono principal() { - return this.exchange.getPrincipal(); - } - - @Override - public Mono> formData() { - return this.exchange.getFormData(); - } - - @Override - public Mono> 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 accept() { - return delegate().getAccept(); - } - - @Override - public List acceptCharset() { - return delegate().getAcceptCharset(); - } - - @Override - public List acceptLanguage() { - return delegate().getAcceptLanguage(); - } - - @Override - public OptionalLong contentLength() { - long value = delegate().getContentLength(); - return (value != -1 ? OptionalLong.of(value) : OptionalLong.empty()); - } - - @Override - public Optional contentType() { - return Optional.ofNullable(delegate().getContentType()); - } - - @Override - public InetSocketAddress host() { - return delegate().getHost(); - } - - @Override - public List range() { - return delegate().getRange(); - } - - @Override - public List header(String headerName) { - List 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(); - } - - } - -} diff --git 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 deleted file mode 100644 index a1d0e4c9..00000000 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/DefaultServerResponse.java +++ /dev/null @@ -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 implements ServerResponse { - - private final ServerWebExchange exchange; - - private final BodyInserter inserter; - - private final Map hints; - - public DefaultServerResponse(ServerWebExchange exchange, - BodyInserter body, Map 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 cookies() { - return response().getCookies(); - } - - @Override - public final Mono writeTo(ServerWebExchange exchange, Context context) { - return this.inserter.insert(exchange.getResponse(), new BodyInserter.Context() { - @Override - public List> messageWriters() { - return context.messageWriters(); - } - - @Override - public Optional serverRequest() { - return Optional.of(exchange.getRequest()); - } - - @Override - public Map 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> messageWriters() { - return this.strategies.messageWriters(); - } - - @Override - public List viewResolvers() { - return this.strategies.viewResolvers(); - } - - } - -}