Polishing
This commit is contained in:
parent
04d5a2951c
commit
137fc48cc2
|
|
@ -55,7 +55,7 @@ public abstract class AbstractClientHttpRequest implements ClientHttpRequest {
|
|||
|
||||
private final MultiValueMap<String, HttpCookie> cookies;
|
||||
|
||||
private AtomicReference<State> state = new AtomicReference<>(State.NEW);
|
||||
private final AtomicReference<State> state = new AtomicReference<>(State.NEW);
|
||||
|
||||
private final List<Supplier<? extends Mono<Void>>> commitActions = new ArrayList<>(4);
|
||||
|
||||
|
|
@ -95,7 +95,7 @@ public abstract class AbstractClientHttpRequest implements ClientHttpRequest {
|
|||
|
||||
@Override
|
||||
public boolean isCommitted() {
|
||||
return this.state.get() != State.NEW;
|
||||
return (this.state.get() != State.NEW);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
/*
|
||||
* Copyright 2002-2016 the original author or authors.
|
||||
* 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.
|
||||
|
|
@ -13,6 +13,7 @@
|
|||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.http.client.reactive;
|
||||
|
||||
import java.net.URI;
|
||||
|
|
@ -37,7 +38,6 @@ public interface ClientHttpConnector {
|
|||
* {@code URI}, then apply the given {@code requestCallback} on the
|
||||
* {@link ClientHttpRequest} once the connection has been established.
|
||||
* <p>Return a publisher of the {@link ClientHttpResponse}.
|
||||
*
|
||||
* @param method the HTTP request method
|
||||
* @param uri the HTTP request URI
|
||||
* @param requestCallback a function that prepares and writes the request,
|
||||
|
|
@ -49,4 +49,4 @@ public interface ClientHttpConnector {
|
|||
Mono<ClientHttpResponse> connect(HttpMethod method, URI uri,
|
||||
Function<? super ClientHttpRequest, Mono<Void>> requestCallback);
|
||||
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@
|
|||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.http.client.reactive;
|
||||
|
||||
import java.net.URI;
|
||||
|
|
@ -42,7 +43,7 @@ public class ClientHttpRequestDecorator implements ClientHttpRequest {
|
|||
|
||||
|
||||
public ClientHttpRequestDecorator(ClientHttpRequest delegate) {
|
||||
Assert.notNull(delegate, "ClientHttpRequest delegate is required.");
|
||||
Assert.notNull(delegate, "Delegate is required");
|
||||
this.delegate = delegate;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@
|
|||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.http.client.reactive;
|
||||
|
||||
import reactor.core.publisher.Flux;
|
||||
|
|
@ -37,7 +38,7 @@ public class ClientHttpResponseDecorator implements ClientHttpResponse {
|
|||
|
||||
|
||||
public ClientHttpResponseDecorator(ClientHttpResponse delegate) {
|
||||
Assert.notNull(delegate, "ClientHttpResponse delegate is required.");
|
||||
Assert.notNull(delegate, "Delegate is required");
|
||||
this.delegate = delegate;
|
||||
}
|
||||
|
||||
|
|
@ -49,7 +50,6 @@ public class ClientHttpResponseDecorator implements ClientHttpResponse {
|
|||
|
||||
// ServerHttpResponse delegation methods...
|
||||
|
||||
|
||||
@Override
|
||||
public HttpStatus getStatusCode() {
|
||||
return this.delegate.getStatusCode();
|
||||
|
|
|
|||
|
|
@ -21,14 +21,14 @@ import java.util.function.Consumer;
|
|||
import java.util.function.Function;
|
||||
|
||||
import reactor.core.publisher.Mono;
|
||||
import reactor.ipc.netty.http.client.HttpClient;
|
||||
import reactor.ipc.netty.http.client.HttpClientOptions;
|
||||
import reactor.ipc.netty.options.ClientOptions;
|
||||
import reactor.ipc.netty.http.client.HttpClient;
|
||||
|
||||
import org.springframework.http.HttpMethod;
|
||||
|
||||
/**
|
||||
* Reactor-Netty implementation of {@link ClientHttpConnector}
|
||||
* Reactor-Netty implementation of {@link ClientHttpConnector}.
|
||||
*
|
||||
* @author Brian Clozel
|
||||
* @see HttpClient
|
||||
|
|
@ -59,7 +59,7 @@ public class ReactorClientHttpConnector implements ClientHttpConnector {
|
|||
public Mono<ClientHttpResponse> connect(HttpMethod method, URI uri,
|
||||
Function<? super ClientHttpRequest, Mono<Void>> requestCallback) {
|
||||
|
||||
return httpClient
|
||||
return this.httpClient
|
||||
.request(io.netty.handler.codec.http.HttpMethod.valueOf(method.name()),
|
||||
uri.toString(),
|
||||
httpClientRequest -> requestCallback
|
||||
|
|
|
|||
|
|
@ -81,14 +81,12 @@ public class ReactorClientHttpRequest extends AbstractClientHttpRequest {
|
|||
|
||||
@Override
|
||||
public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> body) {
|
||||
Publisher<Publisher<ByteBuf>> byteBufs = Flux.from(body).
|
||||
map(ReactorClientHttpRequest::toByteBufs);
|
||||
Publisher<Publisher<ByteBuf>> byteBufs = Flux.from(body).map(ReactorClientHttpRequest::toByteBufs);
|
||||
return doCommit(() -> this.httpRequest.sendGroups(byteBufs).then());
|
||||
}
|
||||
|
||||
private static Publisher<ByteBuf> toByteBufs(Publisher<? extends DataBuffer> dataBuffers) {
|
||||
return Flux.from(dataBuffers).
|
||||
map(NettyDataBufferFactory::toByteBuf);
|
||||
return Flux.from(dataBuffers).map(NettyDataBufferFactory::toByteBuf);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -98,8 +96,7 @@ public class ReactorClientHttpRequest extends AbstractClientHttpRequest {
|
|||
|
||||
@Override
|
||||
protected void applyHeaders() {
|
||||
getHeaders().entrySet()
|
||||
.forEach(e -> this.httpRequest.requestHeaders().set(e.getKey(), e.getValue()));
|
||||
getHeaders().entrySet().forEach(e -> this.httpRequest.requestHeaders().set(e.getKey(), e.getValue()));
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -109,4 +106,4 @@ public class ReactorClientHttpRequest extends AbstractClientHttpRequest {
|
|||
.forEach(this.httpRequest::addCookie);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
/*
|
||||
* Copyright 2002-2016 the original author or authors.
|
||||
* 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.
|
||||
|
|
@ -62,8 +62,7 @@ public class ReactorClientHttpResponse implements ClientHttpResponse {
|
|||
@Override
|
||||
public HttpHeaders getHeaders() {
|
||||
HttpHeaders headers = new HttpHeaders();
|
||||
this.response.responseHeaders()
|
||||
.entries()
|
||||
this.response.responseHeaders().entries()
|
||||
.forEach(e -> headers.add(e.getKey(), e.getValue()));
|
||||
return headers;
|
||||
}
|
||||
|
|
@ -90,12 +89,12 @@ public class ReactorClientHttpResponse implements ClientHttpResponse {
|
|||
return CollectionUtils.unmodifiableMultiValueMap(result);
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "ReactorClientHttpResponse{" +
|
||||
"request=" + this.response.method().name() + " " + this.response.uri() + "," +
|
||||
"status=" + getStatusCode() +
|
||||
'}';
|
||||
"request=[" + this.response.method().name() + " " + this.response.uri() + "]," +
|
||||
"status=" + getStatusCode() + '}';
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue