root
/
spring-framework
mirror of https://github.com/spring-projects/spring-framework.git
1
0
Fork 0
Code Issues Actions 8 Packages Projects Releases Wiki Activity

Polish

This commit is contained in:
Rossen Stoyanchev 2015-11-12 11:52:06 -05:00
parent f1bec5f1e4
commit 141d75791d
6 changed files with 162 additions and 158 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

90
spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/RequestBodyPublisher.java
View File

@ -42,83 +42,66 @@ class RequestBodyPublisher implements Publisher<ByteBuffer> {
private static final AtomicLongFieldUpdater<RequestBodySubscription> DEMAND =
AtomicLongFieldUpdater.newUpdater(RequestBodySubscription.class, "demand");
private final HttpServerExchange exchange;
private Subscriber<? super ByteBuffer> subscriber;
public RequestBodyPublisher(HttpServerExchange exchange) {
Assert.notNull(exchange, "'exchange' is required.");
this.exchange = exchange;
}
@Override
public void subscribe(Subscriber<? super ByteBuffer> s) {
if (s == null) {
public void subscribe(Subscriber<? super ByteBuffer> subscriber) {
if (subscriber == null) {
throw SpecificationExceptions.spec_2_13_exception();
}
if (this.subscriber != null) {
s.onError(new IllegalStateException("Only one subscriber allowed"));
subscriber.onError(new IllegalStateException("Only one subscriber allowed"));
}
this.subscriber = s;
this.subscriber = subscriber;
this.subscriber.onSubscribe(new RequestBodySubscription());
}
private class RequestBodySubscription
implements Subscription, Runnable, ChannelListener<StreamSourceChannel> {
private class RequestBodySubscription implements Subscription, Runnable,
ChannelListener<StreamSourceChannel> {
volatile long demand;
private PooledByteBuffer pooledBuffer;
private StreamSourceChannel channel;
private boolean subscriptionClosed;
private boolean draining;
@Override
public void cancel() {
this.subscriptionClosed = true;
close();
}
@Override
public void request(long n) {
BackpressureUtils.checkRequest(n, subscriber);
if (this.subscriptionClosed) {
return;
}
BackpressureUtils.getAndAdd(DEMAND, this, n);
scheduleNextMessage();
}
private void scheduleNextMessage() {
exchange.dispatch(exchange.isInIoThread() ?
SameThreadExecutor.INSTANCE : exchange.getIoThread(), this);
exchange.dispatch(exchange.isInIoThread() ? SameThreadExecutor.INSTANCE :
exchange.getIoThread(), this);
}
private void doOnNext(ByteBuffer buffer) {
this.draining = false;
buffer.flip();
subscriber.onNext(buffer);
}
private void doOnComplete() {
@Override
public void cancel() {
this.subscriptionClosed = true;
try {
subscriber.onComplete();
}
finally {
close();
}
}
private void doOnError(Throwable t) {
this.subscriptionClosed = true;
try {
subscriber.onError(t);
}
finally {
close();
}
close();
}
private void close() {
@ -137,7 +120,6 @@ class RequestBodyPublisher implements Publisher<ByteBuffer> {
if (this.subscriptionClosed || this.draining) {
return;
}
if (0 == BackpressureUtils.getAndSub(DEMAND, this, 1)) {
return;
}
@ -152,8 +134,7 @@ class RequestBodyPublisher implements Publisher<ByteBuffer> {
return;
}
else {
throw new IllegalStateException(
"Another party already acquired the channel!");
throw new IllegalStateException("Failed to acquire channel!");
}
}
}
@ -198,6 +179,32 @@ class RequestBodyPublisher implements Publisher<ByteBuffer> {
}
}
private void doOnNext(ByteBuffer buffer) {
this.draining = false;
buffer.flip();
subscriber.onNext(buffer);
}
private void doOnComplete() {
this.subscriptionClosed = true;
try {
subscriber.onComplete();
}
finally {
close();
}
}
private void doOnError(Throwable t) {
this.subscriptionClosed = true;
try {
subscriber.onError(t);
}
finally {
close();
}
}
@Override
public void handleEvent(StreamSourceChannel channel) {
if (this.subscriptionClosed) {
@ -237,4 +244,5 @@ class RequestBodyPublisher implements Publisher<ByteBuffer> {
}
}
}
}

62
spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/ResponseBodySubscriber.java
View File

@ -16,11 +16,6 @@
package org.springframework.reactive.web.http.undertow;
import static org.springframework.http.HttpStatus.INTERNAL_SERVER_ERROR;
import static org.xnio.ChannelListeners.closingChannelExceptionHandler;
import static org.xnio.ChannelListeners.flushingChannelListener;
import static org.xnio.IoUtils.safeClose;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Queue;
@ -37,30 +32,43 @@ import org.xnio.ChannelListener;
import org.xnio.channels.StreamSinkChannel;
import reactor.core.subscriber.BaseSubscriber;
import static org.xnio.ChannelListeners.closingChannelExceptionHandler;
import static org.xnio.ChannelListeners.flushingChannelListener;
import static org.xnio.IoUtils.safeClose;
/**
* @author Marek Hawrylczak
* @author Rossen Stoyanchev
*/
class ResponseBodySubscriber extends BaseSubscriber<ByteBuffer>
implements ChannelListener<StreamSinkChannel> {
private static final Log logger = LogFactory.getLog(ResponseBodySubscriber.class);
private final HttpServerExchange exchange;
private final Queue<PooledByteBuffer> buffers;
private final AtomicInteger writing = new AtomicInteger();
private final AtomicBoolean closing = new AtomicBoolean();
private StreamSinkChannel responseChannel;
private Subscription subscription;
private final Queue<PooledByteBuffer> buffers;
private final AtomicInteger writing = new AtomicInteger();
private final AtomicBoolean closing = new AtomicBoolean();
private StreamSinkChannel responseChannel;
public ResponseBodySubscriber(HttpServerExchange exchange) {
this.exchange = exchange;
this.buffers = new ConcurrentLinkedQueue<>();
}
@Override
public void onSubscribe(Subscription s) {
super.onSubscribe(s);
this.subscription = s;
public void onSubscribe(Subscription subscription) {
super.onSubscribe(subscription);
this.subscription = subscription;
this.subscription.request(1);
}
@ -78,6 +86,7 @@ class ResponseBodySubscriber extends BaseSubscriber<ByteBuffer>
do {
c = this.responseChannel.write(buffer);
} while (buffer.hasRemaining() && c > 0);
if (buffer.hasRemaining()) {
this.writing.incrementAndGet();
enqueue(buffer);
@ -102,13 +111,11 @@ class ResponseBodySubscriber extends BaseSubscriber<ByteBuffer>
private void enqueue(ByteBuffer src) {
do {
PooledByteBuffer pooledBuffer =
this.exchange.getConnection().getByteBufferPool().allocate();
ByteBuffer dst = pooledBuffer.getBuffer();
PooledByteBuffer buffer = this.exchange.getConnection().getByteBufferPool().allocate();
ByteBuffer dst = buffer.getBuffer();
copy(dst, src);
dst.flip();
this.buffers.add(pooledBuffer);
this.buffers.add(buffer);
} while (src.remaining() > 0);
}
@ -128,10 +135,12 @@ class ResponseBodySubscriber extends BaseSubscriber<ByteBuffer>
do {
c = channel.write(buffer);
} while (buffer.hasRemaining() && c > 0);
if (!buffer.hasRemaining()) {
safeClose(this.buffers.remove());
}
} while (!this.buffers.isEmpty() && c > 0);
if (!this.buffers.isEmpty()) {
channel.resumeWrites();
}
@ -152,20 +161,17 @@ class ResponseBodySubscriber extends BaseSubscriber<ByteBuffer>
}
@Override
public void onError(Throwable t) {
super.onError(t);
if (!this.exchange.isResponseStarted() &&
this.exchange.getStatusCode() < INTERNAL_SERVER_ERROR.value()) {
this.exchange.setStatusCode(INTERNAL_SERVER_ERROR.value());
public void onError(Throwable ex) {
super.onError(ex);
logger.error("ResponseBodySubscriber error", ex);
if (!this.exchange.isResponseStarted() && this.exchange.getStatusCode() < 500) {
this.exchange.setStatusCode(500);
}
logger.error("ResponseBodySubscriber error", t);
}
@Override
public void onComplete() {
super.onComplete();
if (this.responseChannel != null) {
this.closing.set(true);
closeIfDone();
@ -185,10 +191,8 @@ class ResponseBodySubscriber extends BaseSubscriber<ByteBuffer>
this.responseChannel.shutdownWrites();
if (!this.responseChannel.flush()) {
this.responseChannel.getWriteSetter().set(
flushingChannelListener(
o -> safeClose(this.responseChannel),
closingChannelExceptionHandler()));
this.responseChannel.getWriteSetter().set(flushingChannelListener(
o -> safeClose(this.responseChannel), closingChannelExceptionHandler()));
this.responseChannel.resumeWrites();
}
this.responseChannel = null;

48
spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/RequestHandlerAdapter.java → spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowHttpHandlerAdapter.java
View File

@ -16,41 +16,48 @@
package org.springframework.reactive.web.http.undertow;
import static org.springframework.http.HttpStatus.INTERNAL_SERVER_ERROR;
import org.springframework.http.server.ReactiveServerHttpRequest;
import org.springframework.http.server.ReactiveServerHttpResponse;
import org.springframework.reactive.web.http.HttpHandler;
import org.springframework.util.Assert;
import io.undertow.server.HttpServerExchange;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
/**
* @author Marek Hawrylczak
* @author Rossen Stoyanchev
*/
class RequestHandlerAdapter implements io.undertow.server.HttpHandler {
class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandler {
private final HttpHandler httpHandler;
private static Log logger = LogFactory.getLog(UndertowHttpHandlerAdapter.class);
public RequestHandlerAdapter(HttpHandler httpHandler) {
Assert.notNull(httpHandler, "'httpHandler' is required.");
this.httpHandler = httpHandler;
private final HttpHandler delegate;
public UndertowHttpHandlerAdapter(HttpHandler delegate) {
Assert.notNull(delegate, "'delegate' is required.");
this.delegate = delegate;
}
@Override
public void handleRequest(HttpServerExchange exchange) throws Exception {
RequestBodyPublisher requestBodyPublisher = new RequestBodyPublisher(exchange);
ReactiveServerHttpRequest request =
new UndertowServerHttpRequest(exchange, requestBodyPublisher);
RequestBodyPublisher requestPublisher = new RequestBodyPublisher(exchange);
ReactiveServerHttpRequest request = new UndertowServerHttpRequest(exchange, requestPublisher);
ResponseBodySubscriber responseBodySubscriber = new ResponseBodySubscriber(exchange);
ReactiveServerHttpResponse response =
new UndertowServerHttpResponse(exchange, responseBodySubscriber);
ResponseBodySubscriber responseSubscriber = new ResponseBodySubscriber(exchange);
ReactiveServerHttpResponse response = new UndertowServerHttpResponse(exchange, responseSubscriber);
exchange.dispatch();
this.httpHandler.handle(request, response).subscribe(new Subscriber<Void>() {
this.delegate.handle(request, response).subscribe(new Subscriber<Void>() {
@Override
public void onSubscribe(Subscription subscription) {
subscription.request(Long.MAX_VALUE);
@ -58,14 +65,16 @@ class RequestHandlerAdapter implements io.undertow.server.HttpHandler {
@Override
public void onNext(Void aVoid) {
// no op
}
@Override
public void onError(Throwable t) {
if (!exchange.isResponseStarted() &&
exchange.getStatusCode() < INTERNAL_SERVER_ERROR.value()) {
exchange.setStatusCode(INTERNAL_SERVER_ERROR.value());
public void onError(Throwable ex) {
if (exchange.isResponseStarted() || exchange.getStatusCode() > 500) {
logger.error("Error from request handling. Completing the request.", ex);
}
else {
exchange.setStatusCode(500);
}
exchange.endExchange();
}
@ -76,4 +85,5 @@ class RequestHandlerAdapter implements io.undertow.server.HttpHandler {
}
});
}
}

32
spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowHttpServer.java
View File

@ -27,44 +27,42 @@ import io.undertow.server.HttpHandler;
/**
* @author Marek Hawrylczak
*/
public class UndertowHttpServer extends HttpServerSupport
implements InitializingBean, HttpServer {
public class UndertowHttpServer extends HttpServerSupport implements InitializingBean, HttpServer {
private Undertow undertowServer;
private Undertow server;
private boolean running;
@Override
public void afterPropertiesSet() throws Exception {
Assert.notNull(getHttpHandler());
HttpHandler handler = new RequestHandlerAdapter(getHttpHandler());
this.undertowServer = Undertow.builder()
.addHttpListener(getPort() != -1 ? getPort() : 8080, "localhost")
.setHandler(handler)
.build();
HttpHandler handler = new UndertowHttpHandlerAdapter(getHttpHandler());
int port = (getPort() != -1 ? getPort() : 8080);
this.server = Undertow.builder().addHttpListener(port, "localhost")
.setHandler(handler).build();
}
@Override
public void start() {
if (!running) {
this.undertowServer.start();
running = true;
if (!this.running) {
this.server.start();
this.running = true;
}
}
@Override
public void stop() {
if (running) {
this.undertowServer.stop();
running = false;
if (this.running) {
this.server.stop();
this.running = false;
}
}
@Override
public boolean isRunning() {
return running;
return this.running;
}
}

35
spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowServerHttpRequest.java
View File

@ -20,37 +20,32 @@ import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.server.ReactiveServerHttpRequest;
import org.springframework.util.StringUtils;
import io.undertow.server.HttpServerExchange;
import io.undertow.util.HeaderValues;
import org.reactivestreams.Publisher;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.server.ReactiveServerHttpRequest;
/**
* @author Marek Hawrylczak
* @author Rossen Stoyanchev
*/
class UndertowServerHttpRequest implements ReactiveServerHttpRequest {
private final HttpServerExchange exchange;
private final Publisher<ByteBuffer> requestBodyPublisher;
private final Publisher<ByteBuffer> body;
private HttpHeaders headers;
public UndertowServerHttpRequest(HttpServerExchange exchange,
Publisher<ByteBuffer> requestBodyPublisher) {
public UndertowServerHttpRequest(HttpServerExchange exchange, Publisher<ByteBuffer> body) {
this.exchange = exchange;
this.requestBodyPublisher = requestBodyPublisher;
this.body = body;
}
@Override
public Publisher<ByteBuffer> getBody() {
return this.requestBodyPublisher;
}
@Override
public HttpMethod getMethod() {
@ -60,11 +55,9 @@ class UndertowServerHttpRequest implements ReactiveServerHttpRequest {
@Override
public URI getURI() {
try {
StringBuilder uri = new StringBuilder(this.exchange.getRequestPath());
if (StringUtils.hasLength(this.exchange.getQueryString())) {
uri.append('?').append(this.exchange.getQueryString());
}
return new URI(uri.toString());
return new URI(this.exchange.getRequestScheme(), null, this.exchange.getHostName(),
this.exchange.getHostPort(), this.exchange.getRequestURI(),
this.exchange.getQueryString(), null);
}
catch (URISyntaxException ex) {
throw new IllegalStateException("Could not get URI: " + ex.getMessage(), ex);
@ -83,4 +76,10 @@ class UndertowServerHttpRequest implements ReactiveServerHttpRequest {
}
return this.headers;
}
@Override
public Publisher<ByteBuffer> getBody() {
return this.body;
}
}

53
spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowServerHttpResponse.java
View File

@ -23,65 +23,50 @@ import java.util.Map;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.server.ReactiveServerHttpResponse;
import org.springframework.util.Assert;
import io.undertow.server.HttpServerExchange;
import io.undertow.util.HttpString;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscription;
import reactor.rx.Streams;
/**
* @author Marek Hawrylczak
* @author Rossen Stoyanchev
*/
class UndertowServerHttpResponse implements ReactiveServerHttpResponse {
private final HttpServerExchange exchange;
private final HttpHeaders headers;
private final ResponseBodySubscriber responseBodySubscriber;
private final HttpServerExchange exchange;
private final ResponseBodySubscriber bodySubscriber;
private final HttpHeaders headers = new HttpHeaders();
private boolean headersWritten = false;
public UndertowServerHttpResponse(HttpServerExchange exchange,
ResponseBodySubscriber responseBodySubscriber) {
public UndertowServerHttpResponse(HttpServerExchange exchange, ResponseBodySubscriber body) {
this.exchange = exchange;
this.responseBodySubscriber = responseBodySubscriber;
this.headers = new HttpHeaders();
this.bodySubscriber = body;
}
@Override
public void setStatusCode(HttpStatus status) {
Assert.notNull(status);
this.exchange.setStatusCode(status.value());
}
@Override
public Publisher<Void> setBody(Publisher<ByteBuffer> contentPublisher) {
applyHeaders();
return s -> s.onSubscribe(new Subscription() {
@Override
public void request(long n) {
Streams.wrap(contentPublisher)
.finallyDo(byteBufferSignal -> {
if (byteBufferSignal.isOnComplete()) {
s.onComplete();
}
else {
s.onError(byteBufferSignal.getThrowable());
}
}
).subscribe(responseBodySubscriber);
}
@Override
public void cancel() {
}
});
@Override
public Publisher<Void> setBody(Publisher<ByteBuffer> bodyPublisher) {
applyHeaders();
return (subscriber -> bodyPublisher.subscribe(bodySubscriber));
}
@Override
public HttpHeaders getHeaders() {
return (this.headersWritten ?
HttpHeaders.readOnlyHttpHeaders(this.headers) : this.headers);
return (this.headersWritten ? HttpHeaders.readOnlyHttpHeaders(this.headers) : this.headers);
}
@Override
@ -102,12 +87,12 @@ class UndertowServerHttpResponse implements ReactiveServerHttpResponse {
private void applyHeaders() {
if (!this.headersWritten) {
for (Map.Entry<String, List<String>> entry : this.headers.entrySet()) {
String headerName = entry.getKey();
this.exchange.getResponseHeaders()
.addAll(HttpString.tryFromString(headerName), entry.getValue());
HttpString headerName = HttpString.tryFromString(entry.getKey());
this.exchange.getResponseHeaders().addAll(headerName, entry.getValue());
}
this.headersWritten = true;
}
}
}