Reactive setup refinements

This commit is contained in:
Juergen Hoeller
2016-09-21 11:24:20 +02:00
parent 578af59f0c
commit 89717e1783
10 changed files with 118 additions and 122 deletions

View File

@@ -37,13 +37,12 @@ public class ReactorHttpHandlerAdapter implements Function<HttpChannel, Mono<Voi
private static Log logger = LogFactory.getLog(ReactorHttpHandlerAdapter.class);
private final HttpHandler httpHandler;
private final HttpHandler delegate;
public ReactorHttpHandlerAdapter(HttpHandler httpHandler) {
Assert.notNull(httpHandler, "HttpHandler is required");
this.httpHandler = httpHandler;
public ReactorHttpHandlerAdapter(HttpHandler delegate) {
Assert.notNull(delegate, "HttpHandler delegate is required");
this.delegate = delegate;
}
@@ -52,7 +51,8 @@ public class ReactorHttpHandlerAdapter implements Function<HttpChannel, Mono<Voi
NettyDataBufferFactory bufferFactory = new NettyDataBufferFactory(channel.delegate().alloc());
ReactorServerHttpRequest adaptedRequest = new ReactorServerHttpRequest(channel, bufferFactory);
ReactorServerHttpResponse adaptedResponse = new ReactorServerHttpResponse(channel, bufferFactory);
return this.httpHandler.handle(adaptedRequest, adaptedResponse)
return this.delegate.handle(adaptedRequest, adaptedResponse)
.otherwise(ex -> {
logger.debug("Could not complete request", ex);
channel.status(HttpResponseStatus.INTERNAL_SERVER_ERROR);

View File

@@ -21,7 +21,6 @@ import io.netty.handler.codec.http.HttpResponseStatus;
import io.reactivex.netty.protocol.http.server.HttpServerRequest;
import io.reactivex.netty.protocol.http.server.HttpServerResponse;
import io.reactivex.netty.protocol.http.server.RequestHandler;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;
@@ -42,13 +41,12 @@ public class RxNettyHttpHandlerAdapter implements RequestHandler<ByteBuf, ByteBu
private static Log logger = LogFactory.getLog(RxNettyHttpHandlerAdapter.class);
private final HttpHandler httpHandler;
private final HttpHandler delegate;
public RxNettyHttpHandlerAdapter(HttpHandler httpHandler) {
Assert.notNull(httpHandler, "'httpHandler' is required");
this.httpHandler = httpHandler;
public RxNettyHttpHandlerAdapter(HttpHandler delegate) {
Assert.notNull(delegate, "HttpHandler delegate is required");
this.delegate = delegate;
}
@@ -57,13 +55,15 @@ public class RxNettyHttpHandlerAdapter implements RequestHandler<ByteBuf, ByteBu
NettyDataBufferFactory bufferFactory = new NettyDataBufferFactory(response.unsafeNettyChannel().alloc());
RxNettyServerHttpRequest adaptedRequest = new RxNettyServerHttpRequest(request, bufferFactory);
RxNettyServerHttpResponse adaptedResponse = new RxNettyServerHttpResponse(response, bufferFactory);
Publisher<Void> result = this.httpHandler.handle(adaptedRequest, adaptedResponse)
Publisher<Void> result = this.delegate.handle(adaptedRequest, adaptedResponse)
.otherwise(ex -> {
logger.debug("Could not complete request", ex);
response.setStatus(HttpResponseStatus.INTERNAL_SERVER_ERROR);
return Mono.empty();
})
.doOnSuccess(aVoid -> logger.debug("Successfully completed request"));
return RxJava1Adapter.publisherToObservable(result);
}

View File

@@ -45,37 +45,31 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle
public UndertowHttpHandlerAdapter(HttpHandler delegate) {
Assert.notNull(delegate, "'delegate' is required");
Assert.notNull(delegate, "HttpHandler delegate is required");
this.delegate = delegate;
}
public void setDataBufferFactory(DataBufferFactory dataBufferFactory) {
Assert.notNull(dataBufferFactory, "'dataBufferFactory' must not be null");
Assert.notNull(dataBufferFactory, "DataBufferFactory must not be null");
this.dataBufferFactory = dataBufferFactory;
}
@Override
public void handleRequest(HttpServerExchange exchange) throws Exception {
ServerHttpRequest request =
new UndertowServerHttpRequest(exchange, this.dataBufferFactory);
ServerHttpResponse response =
new UndertowServerHttpResponse(exchange, this.dataBufferFactory);
ServerHttpRequest request = new UndertowServerHttpRequest(exchange, this.dataBufferFactory);
ServerHttpResponse response = new UndertowServerHttpResponse(exchange, this.dataBufferFactory);
this.delegate.handle(request, response).subscribe(new Subscriber<Void>() {
@Override
public void onSubscribe(Subscription subscription) {
subscription.request(Long.MAX_VALUE);
}
@Override
public void onNext(Void aVoid) {
// no op
}
@Override
public void onError(Throwable ex) {
logger.debug("Could not complete request", ex);
@@ -84,7 +78,6 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle
}
exchange.endExchange();
}
@Override
public void onComplete() {
logger.debug("Successfully completed request");
@@ -93,4 +86,4 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle
});
}
}
}