Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Decouple ExchangeResult from WiretapRequest/Response

Browse Source 
This commit decouples ExchangeResult from knowledge about
WiretapClientHttpRequest/Response both of which are now private to
WiretapConnector.

ExchangeResult takes ClientHttpRequest/Response instead along with
promises for the serialized request and response content in the form
of MonoProcessor<byte[]>.

This sets up the possibility for an ExchangeResult to be created
outside and independent of WebTestClient, should we decide to make its
constructor public.

Issue: SPR-16101
This commit is contained in:
Rossen Stoyanchev
2017-10-25 17:26:36 -04:00
parent 7e8c8f0b76
commit a982123ed5
9 changed files with 220 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
spring-test/src/main/java/org/springframework/test/web/reactive/server/DefaultWebTestClient.java
View File

@@ -281,50 +281,50 @@ class DefaultWebTestClient implements WebTestClient {
public ResponseSpec exchange() {
ClientResponse clientResponse = this.bodySpec.exchange().block(getTimeout());
Assert.state(clientResponse != null, "No ClientResponse");
ExchangeResult exchangeResult = wiretapConnector.claimRequest(this.requestId);
return new DefaultResponseSpec(exchangeResult, clientResponse, this.uriTemplate, getTimeout());
WiretapConnector.Info info = wiretapConnector.claimRequest(this.requestId);
return new DefaultResponseSpec(info, clientResponse, this.uriTemplate, getTimeout());
}
}
private static class DefaultResponseSpec implements ResponseSpec {
private final ExchangeResult result;
private final ExchangeResult exchangeResult;
private final ClientResponse response;
private final Duration timeout;
DefaultResponseSpec(ExchangeResult result, ClientResponse response,
DefaultResponseSpec(WiretapConnector.Info wiretapInfo, ClientResponse response,
@Nullable String uriTemplate, Duration timeout) {
this.result = new ExchangeResult(result, uriTemplate);
this.exchangeResult = wiretapInfo.createExchangeResult(uriTemplate);
this.response = response;
this.timeout = timeout;
}
@Override
public StatusAssertions expectStatus() {
return new StatusAssertions(this.result, this);
return new StatusAssertions(this.exchangeResult, this);
}
@Override
public HeaderAssertions expectHeader() {
return new HeaderAssertions(this.result, this);
return new HeaderAssertions(this.exchangeResult, this);
}
@Override
public <B> BodySpec<B, ?> expectBody(Class<B> bodyType) {
B body = this.response.bodyToMono(bodyType).block(this.timeout);
EntityExchangeResult<B> entityResult = new EntityExchangeResult<>(this.result, body);
EntityExchangeResult<B> entityResult = new EntityExchangeResult<>(this.exchangeResult, body);
return new DefaultBodySpec<>(entityResult);
}
@Override
public <B> BodySpec<B, ?> expectBody(ParameterizedTypeReference<B> bodyType) {
B body = this.response.bodyToMono(bodyType).block(this.timeout);
EntityExchangeResult<B> entityResult = new EntityExchangeResult<>(this.result, body);
EntityExchangeResult<B> entityResult = new EntityExchangeResult<>(this.exchangeResult, body);
return new DefaultBodySpec<>(entityResult);
}
@@ -341,7 +341,7 @@ class DefaultWebTestClient implements WebTestClient {
private <E> ListBodySpec<E> getListBodySpec(Flux<E> flux) {
List<E> body = flux.collectList().block(this.timeout);
EntityExchangeResult<List<E>> entityResult = new EntityExchangeResult<>(this.result, body);
EntityExchangeResult<List<E>> entityResult = new EntityExchangeResult<>(this.exchangeResult, body);
return new DefaultListBodySpec<>(entityResult);
}
@@ -349,20 +349,20 @@ class DefaultWebTestClient implements WebTestClient {
public BodyContentSpec expectBody() {
ByteArrayResource resource = this.response.bodyToMono(ByteArrayResource.class).block(this.timeout);
byte[] body = (resource != null ? resource.getByteArray() : null);
EntityExchangeResult<byte[]> entityResult = new EntityExchangeResult<>(this.result, body);
EntityExchangeResult<byte[]> entityResult = new EntityExchangeResult<>(this.exchangeResult, body);
return new DefaultBodyContentSpec(entityResult);
}
@Override
public <T> FluxExchangeResult<T> returnResult(Class<T> elementType) {
Flux<T> body = this.response.bodyToFlux(elementType);
return new FluxExchangeResult<>(this.result, body, this.timeout);
return new FluxExchangeResult<>(this.exchangeResult, body, this.timeout);
}
@Override
public <T> FluxExchangeResult<T> returnResult(ParameterizedTypeReference<T> elementType) {
Flux<T> body = this.response.bodyToFlux(elementType);
return new FluxExchangeResult<>(this.result, body, this.timeout);
return new FluxExchangeResult<>(this.exchangeResult, body, this.timeout);
}
}

52
spring-test/src/main/java/org/springframework/test/web/reactive/server/ExchangeResult.java
View File

@@ -31,6 +31,8 @@ import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseCookie;
import org.springframework.http.client.reactive.ClientHttpRequest;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.MultiValueMap;
@@ -58,31 +60,41 @@ public class ExchangeResult {
MediaType.parseMediaType("text/*"), MediaType.APPLICATION_FORM_URLENCODED);
private final WiretapClientHttpRequest request;
private final ClientHttpRequest request;
private final WiretapClientHttpResponse response;
private final ClientHttpResponse response;
private final MonoProcessor<byte[]> requestBody;
private final MonoProcessor<byte[]> responseBody;
@Nullable
private final String uriTemplate;
/**
* Constructor to use after the server response is first received in the
* {@link WiretapConnector} and the {@code ClientHttpResponse} created.
* Create an instance with an HTTP request and response along with promises
* for the serialized request and response body content.
*
* @param request the HTTP request
* @param response the HTTP response
* @param requestBody capture of serialized request body content
* @param responseBody capture of serialized response body content
* @param uriTemplate the URI template used to set up the request, if any
*/
ExchangeResult(WiretapClientHttpRequest request, WiretapClientHttpResponse response) {
ExchangeResult(ClientHttpRequest request, ClientHttpResponse response,
MonoProcessor<byte[]> requestBody, MonoProcessor<byte[]> responseBody,
@Nullable String uriTemplate) {
Assert.notNull(request, "ClientHttpRequest is required");
Assert.notNull(response, "ClientHttpResponse is required");
Assert.notNull(requestBody, "'requestBody' is required");
Assert.notNull(responseBody, "'responseBody' is required");
this.request = request;
this.response = response;
this.uriTemplate = null;
}
/**
* Constructor to copy the from the yet undecoded ExchangeResult with extra
* information to expose such as the original URI template used, if any.
*/
ExchangeResult(ExchangeResult other, @Nullable String uriTemplate) {
this.request = other.request;
this.response = other.response;
this.requestBody = requestBody;
this.responseBody = responseBody;
this.uriTemplate = uriTemplate;
}
@@ -92,6 +104,8 @@ public class ExchangeResult {
ExchangeResult(ExchangeResult other) {
this.request = other.request;
this.response = other.response;
this.requestBody = other.requestBody;
this.responseBody = other.responseBody;
this.uriTemplate = other.uriTemplate;
}
@@ -131,7 +145,7 @@ public class ExchangeResult {
*/
@Nullable
public byte[] getRequestBodyContent() {
MonoProcessor<byte[]> body = this.request.getRecordedContent();
MonoProcessor<byte[]> body = this.requestBody;
Assert.isTrue(body.isTerminated(), "Request body incomplete.");
return body.block(Duration.ZERO);
}
@@ -164,7 +178,7 @@ public class ExchangeResult {
*/
@Nullable
public byte[] getResponseBodyContent() {
MonoProcessor<byte[]> body = this.response.getRecordedContent();
MonoProcessor<byte[]> body = this.responseBody;
Assert.state(body.isTerminated(), "Response body incomplete");
return body.block(Duration.ZERO);
}
@@ -191,12 +205,12 @@ public class ExchangeResult {
"> " + getMethod() + " " + getUrl() + "\n" +
"> " + formatHeaders(getRequestHeaders(), "\n> ") + "\n" +
"\n" +
formatBody(getRequestHeaders().getContentType(), this.request.getRecordedContent()) + "\n" +
formatBody(getRequestHeaders().getContentType(), this.requestBody) + "\n" +
"\n" +
"< " + getStatus() + " " + getStatusReason() + "\n" +
"< " + formatHeaders(getResponseHeaders(), "\n< ") + "\n" +
"\n" +
formatBody(getResponseHeaders().getContentType(), this.response.getRecordedContent()) +"\n";
formatBody(getResponseHeaders().getContentType(), this.responseBody) +"\n";
}
private String getStatusReason() {

105
spring-test/src/main/java/org/springframework/test/web/reactive/server/WiretapClientHttpRequest.java
View File

@@ -1,105 +0,0 @@
/*
* Copyright 2002-2017 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.test.web.reactive.server;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.http.client.reactive.ClientHttpRequest;
import org.springframework.http.client.reactive.ClientHttpRequestDecorator;
/**
* Client HTTP request decorator that intercepts and saves content written to
* the server.
*
* @author Rossen Stoyanchev
* @since 5.0
*/
class WiretapClientHttpRequest extends ClientHttpRequestDecorator {
private static final DataBufferFactory bufferFactory = new DefaultDataBufferFactory();
private final DataBuffer buffer;
private final MonoProcessor<byte[]> body = MonoProcessor.create();
public WiretapClientHttpRequest(ClientHttpRequest delegate) {
super(delegate);
this.buffer = bufferFactory.allocateBuffer();
}
/**
* Return a "promise" with the request body content written to the server.
*/
public MonoProcessor<byte[]> getRecordedContent() {
return this.body;
}
@Override
public Mono<Void> writeWith(Publisher<? extends DataBuffer> publisher) {
return super.writeWith(
Flux.from(publisher)
.doOnNext(this::handleOnNext)
.doOnError(this::handleError)
.doOnCancel(this::handleOnComplete)
.doOnComplete(this::handleOnComplete));
}
@Override
public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> publisher) {
return super.writeAndFlushWith(
Flux.from(publisher)
.map(p -> Flux.from(p).doOnNext(this::handleOnNext).doOnError(this::handleError))
.doOnError(this::handleError)
.doOnCancel(this::handleOnComplete)
.doOnComplete(this::handleOnComplete));
}
@Override
public Mono<Void> setComplete() {
handleOnComplete();
return super.setComplete();
}
private void handleOnNext(DataBuffer buffer) {
this.buffer.write(buffer);
}
private void handleError(Throwable ex) {
if (!this.body.isTerminated()) {
this.body.onError(ex);
}
}
private void handleOnComplete() {
if (!this.body.isTerminated()) {
byte[] bytes = new byte[this.buffer.readableByteCount()];
this.buffer.read(bytes);
this.body.onNext(bytes);
}
}
}

75
spring-test/src/main/java/org/springframework/test/web/reactive/server/WiretapClientHttpResponse.java
View File

@@ -1,75 +0,0 @@
/*
* Copyright 2002-2017 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.test.web.reactive.server;
import reactor.core.publisher.Flux;
import reactor.core.publisher.MonoProcessor;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.http.client.reactive.ClientHttpResponseDecorator;
/**
* Client HTTP response decorator that intercepts and saves the content read
* from the server.
*
* @author Rossen Stoyanchev
* @since 5.0
*/
class WiretapClientHttpResponse extends ClientHttpResponseDecorator {
private static final DataBufferFactory bufferFactory = new DefaultDataBufferFactory();
private final DataBuffer buffer;
private final MonoProcessor<byte[]> body = MonoProcessor.create();
public WiretapClientHttpResponse(ClientHttpResponse delegate) {
super(delegate);
this.buffer = bufferFactory.allocateBuffer();
}
/**
* Return a "promise" with the response body content read from the server.
*/
public MonoProcessor<byte[]> getRecordedContent() {
return this.body;
}
@Override
public Flux<DataBuffer> getBody() {
return super.getBody()
.doOnNext(buffer::write)
.doOnError(body::onError)
.doOnCancel(this::handleOnComplete)
.doOnComplete(this::handleOnComplete);
}
private void handleOnComplete() {
if (!this.body.isTerminated()) {
byte[] bytes = new byte[this.buffer.readableByteCount()];
this.buffer.read(bytes);
this.body.onNext(bytes);
}
}
}

167
spring-test/src/main/java/org/springframework/test/web/reactive/server/WiretapConnector.java
View File

@@ -22,12 +22,21 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.http.HttpMethod;
import org.springframework.http.client.reactive.ClientHttpConnector;
import org.springframework.http.client.reactive.ClientHttpRequest;
import org.springframework.http.client.reactive.ClientHttpRequestDecorator;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.http.client.reactive.ClientHttpResponseDecorator;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
@@ -41,9 +50,12 @@ import org.springframework.util.Assert;
*/
class WiretapConnector implements ClientHttpConnector {
private static final DataBufferFactory bufferFactory = new DefaultDataBufferFactory();
private final ClientHttpConnector delegate;
private final Map<String, ExchangeResult> exchanges = new ConcurrentHashMap<>();
private final Map<String, Info> exchanges = new ConcurrentHashMap<>();
WiretapConnector(ClientHttpConnector delegate) {
@@ -65,22 +77,157 @@ class WiretapConnector implements ClientHttpConnector {
})
.map(response -> {
WiretapClientHttpRequest wrappedRequest = requestRef.get();
String requestId = wrappedRequest.getHeaders().getFirst(WebTestClient.WEBTESTCLIENT_REQUEST_ID);
Assert.state(requestId != null, () -> "No \"" + WebTestClient.WEBTESTCLIENT_REQUEST_ID + "\" header");
String header = WebTestClient.WEBTESTCLIENT_REQUEST_ID;
String requestId = wrappedRequest.getHeaders().getFirst(header);
Assert.state(requestId != null, () -> "No \"" + header + "\" header");
WiretapClientHttpResponse wrappedResponse = new WiretapClientHttpResponse(response);
ExchangeResult result = new ExchangeResult(wrappedRequest, wrappedResponse);
this.exchanges.put(requestId, result);
this.exchanges.put(requestId, new Info(wrappedRequest, wrappedResponse));
return wrappedResponse;
});
}
/**
* Retrieve the {@code ExchangeResult} for the given "request-id" header value.
* Retrieve the {@link Info} for the given "request-id" header value.
*/
public ExchangeResult claimRequest(String requestId) {
ExchangeResult result = this.exchanges.remove(requestId);
Assert.state(result != null, () -> "No match for " + WebTestClient.WEBTESTCLIENT_REQUEST_ID + "=" + requestId);
return result;
public Info claimRequest(String requestId) {
Info info = this.exchanges.remove(requestId);
Assert.state(info != null, () -> {
String header = WebTestClient.WEBTESTCLIENT_REQUEST_ID;
return "No match for " + header + "=" + requestId;
});
return info;
}
class Info {
private final WiretapClientHttpRequest request;
private final WiretapClientHttpResponse response;
public Info(WiretapClientHttpRequest request, WiretapClientHttpResponse response) {
this.request = request;
this.response = response;
}
public ExchangeResult createExchangeResult(@Nullable String uriTemplate) {
return new ExchangeResult(this.request, this.response,
this.request.getContent(), this.response.getContent(), uriTemplate);
}
}
/**
* ClientHttpRequestDecorator that intercepts and saves the request body.
*/
private static class WiretapClientHttpRequest extends ClientHttpRequestDecorator {
private final DataBuffer buffer;
private final MonoProcessor<byte[]> body = MonoProcessor.create();
public WiretapClientHttpRequest(ClientHttpRequest delegate) {
super(delegate);
this.buffer = bufferFactory.allocateBuffer();
}
/**
* Return a "promise" with the request body content written to the server.
*/
public MonoProcessor<byte[]> getContent() {
return this.body;
}
@Override
public Mono<Void> writeWith(Publisher<? extends DataBuffer> publisher) {
return super.writeWith(
Flux.from(publisher)
.doOnNext(this::handleOnNext)
.doOnError(this::handleError)
.doOnCancel(this::handleOnComplete)
.doOnComplete(this::handleOnComplete));
}
@Override
public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> publisher) {
return super.writeAndFlushWith(
Flux.from(publisher)
.map(p -> Flux.from(p).doOnNext(this::handleOnNext).doOnError(this::handleError))
.doOnError(this::handleError)
.doOnCancel(this::handleOnComplete)
.doOnComplete(this::handleOnComplete));
}
@Override
public Mono<Void> setComplete() {
handleOnComplete();
return super.setComplete();
}
private void handleOnNext(DataBuffer buffer) {
this.buffer.write(buffer);
}
private void handleError(Throwable ex) {
if (!this.body.isTerminated()) {
this.body.onError(ex);
}
}
private void handleOnComplete() {
if (!this.body.isTerminated()) {
byte[] bytes = new byte[this.buffer.readableByteCount()];
this.buffer.read(bytes);
this.body.onNext(bytes);
}
}
}
/**
* ClientHttpResponseDecorator that intercepts and saves the response body.
*/
private static class WiretapClientHttpResponse extends ClientHttpResponseDecorator {
private final DataBuffer buffer;
private final MonoProcessor<byte[]> body = MonoProcessor.create();
public WiretapClientHttpResponse(ClientHttpResponse delegate) {
super(delegate);
this.buffer = bufferFactory.allocateBuffer();
}
/**
* Return a "promise" with the response body content read from the server.
*/
public MonoProcessor<byte[]> getContent() {
return this.body;
}
@Override
public Flux<DataBuffer> getBody() {
return super.getBody()
.doOnNext(buffer::write)
.doOnError(body::onError)
.doOnCancel(this::handleOnComplete)
.doOnComplete(this::handleOnComplete);
}
private void handleOnComplete() {
if (!this.body.isTerminated()) {
byte[] bytes = new byte[this.buffer.readableByteCount()];
this.buffer.read(bytes);
this.body.onNext(bytes);
}
}
}
}