diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/RequestBodyPublisher.java b/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/RequestBodyPublisher.java index ffa21597dd..db0a0e89e7 100644 --- a/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/RequestBodyPublisher.java +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/RequestBodyPublisher.java @@ -42,83 +42,66 @@ class RequestBodyPublisher implements Publisher { private static final AtomicLongFieldUpdater DEMAND = AtomicLongFieldUpdater.newUpdater(RequestBodySubscription.class, "demand"); + + private final HttpServerExchange exchange; + private Subscriber subscriber; + public RequestBodyPublisher(HttpServerExchange exchange) { Assert.notNull(exchange, "'exchange' is required."); this.exchange = exchange; } + @Override - public void subscribe(Subscriber s) { - if (s == null) { + public void subscribe(Subscriber 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 { + + private class RequestBodySubscription implements Subscription, Runnable, + ChannelListener { 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 { if (this.subscriptionClosed || this.draining) { return; } - if (0 == BackpressureUtils.getAndSub(DEMAND, this, 1)) { return; } @@ -152,8 +134,7 @@ class RequestBodyPublisher implements Publisher { 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 { } } + 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 { } } } + } diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/ResponseBodySubscriber.java b/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/ResponseBodySubscriber.java index ae42aa1a2d..b78a687fa6 100644 --- a/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/ResponseBodySubscriber.java +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/ResponseBodySubscriber.java @@ -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 implements ChannelListener { private static final Log logger = LogFactory.getLog(ResponseBodySubscriber.class); + private final HttpServerExchange exchange; - private final Queue buffers; - private final AtomicInteger writing = new AtomicInteger(); - private final AtomicBoolean closing = new AtomicBoolean(); - private StreamSinkChannel responseChannel; + private Subscription subscription; + private final Queue 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 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 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 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 } @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 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; diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/RequestHandlerAdapter.java b/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowHttpHandlerAdapter.java similarity index 56% rename from spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/RequestHandlerAdapter.java rename to spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowHttpHandlerAdapter.java index b87f126818..92ad9ff3f0 100644 --- a/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/RequestHandlerAdapter.java +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowHttpHandlerAdapter.java @@ -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() { + + this.delegate.handle(request, response).subscribe(new Subscriber() { + @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 { } }); } + } diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowHttpServer.java b/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowHttpServer.java index 6a9a747c18..b55fa2201f 100644 --- a/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowHttpServer.java +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowHttpServer.java @@ -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; } + } diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowServerHttpRequest.java b/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowServerHttpRequest.java index 4e83006676..6f856595f3 100644 --- a/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowServerHttpRequest.java +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowServerHttpRequest.java @@ -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 requestBodyPublisher; + private final Publisher body; private HttpHeaders headers; - public UndertowServerHttpRequest(HttpServerExchange exchange, - Publisher requestBodyPublisher) { + public UndertowServerHttpRequest(HttpServerExchange exchange, Publisher body) { this.exchange = exchange; - this.requestBodyPublisher = requestBodyPublisher; + this.body = body; } - @Override - public Publisher 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 getBody() { + return this.body; + } + } diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowServerHttpResponse.java b/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowServerHttpResponse.java index b3bc0fefe9..5efe890de1 100644 --- a/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowServerHttpResponse.java +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowServerHttpResponse.java @@ -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 setBody(Publisher 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 setBody(Publisher 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> 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; } } + }