Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Simplify HttpTrafficHandler#write #3376

Merged
merged 1 commit into from
Jul 29, 2024
+133 −47
Merged

Simplify HttpTrafficHandler#write #3376

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
180 changes: 133 additions & 47 deletions reactor-netty-http/src/main/java/reactor/netty/http/server/HttpTrafficHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,10 @@
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.DecoderResult;
import io.netty.handler.codec.DecoderResultProvider;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.DefaultHttpContent;
import io.netty.handler.codec.http.DefaultHttpResponse;
import io.netty.handler.codec.http.DefaultLastHttpContent;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpObject;
import io.netty.handler.codec.http.HttpRequest;
Expand Down Expand Up @@ -56,6 +60,7 @@
import static io.netty.handler.codec.http.HttpUtil.isKeepAlive;
import static io.netty.handler.codec.http.HttpUtil.isTransferEncodingChunked;
import static io.netty.handler.codec.http.HttpUtil.setKeepAlive;
import static io.netty.handler.codec.http.LastHttpContent.EMPTY_LAST_CONTENT;
import static reactor.netty.ReactorNetty.format;

/**
Expand Down Expand Up @@ -360,8 +365,28 @@ void doPipeline(ChannelHandlerContext ctx, Object msg) {
@Override
@SuppressWarnings("FutureReturnValueIgnored")
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
Class<?> msgClass = msg.getClass();
// modify message on way out to add headers if needed
if (msg instanceof HttpResponse) {
if (msgClass == DefaultHttpResponse.class) {
handleDefaultHttpResponse((DefaultHttpResponse) msg, promise);
return;
}
else if (msgClass == DefaultFullHttpResponse.class) {
if (handleDefaultFullHttpResponse((DefaultFullHttpResponse) msg, promise)) {
return;
}
handleLastHttpContent(msg, promise);
return;
}
else if (msg == EMPTY_LAST_CONTENT || msgClass == DefaultLastHttpContent.class) {
handleLastHttpContent(msg, promise);
return;
}
else if (msgClass == DefaultHttpContent.class) {
handleDefaultHttContent((DefaultHttpContent) msg, promise);
return;
}
else if (msg instanceof HttpResponse) {
final HttpResponse response = (HttpResponse) msg;
nonInformationalResponse = !isInformational(response);
// Assume the response writer knows if they can persist or not and sets isKeepAlive on the response
Expand All @@ -383,52 +408,7 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
}
}
if (msg instanceof LastHttpContent) {
finalizingResponse = true;

if (LAST_FLUSH_WHEN_NO_READ) {
needsFlush = !read;
}

if (!shouldKeepAlive()) {
if (HttpServerOperations.log.isDebugEnabled()) {
HttpServerOperations.log.debug(format(ctx.channel(), "Detected non persistent http " +
"connection, preparing to close. Pending responses count: {}"),
pendingResponses);
}
ctx.write(msg, promise.unvoid())
.addListener(ChannelFutureListener.CLOSE);
return;
}

ctx.write(msg, promise);

if (!persistentConnection) {
return;
}

if (nonInformationalResponse) {
nonInformationalResponse = false;
pendingResponses -= 1;
if (HttpServerOperations.log.isDebugEnabled()) {
HttpServerOperations.log.debug(format(ctx.channel(), "Decreasing pending responses count: {}"),
pendingResponses);
}
}

if (pipelined != null && !pipelined.isEmpty()) {
if (HttpServerOperations.log.isDebugEnabled()) {
HttpServerOperations.log.debug(format(ctx.channel(), "Draining next pipelined " +
"HTTP request, pending responses count: {}, queued: {}"),
pendingResponses, pipelined.size());
}
ctx.executor()
.execute(this);
}
else {
IdleTimeoutHandler.addIdleTimeoutHandler(ctx.pipeline(), idleTimeout);

ctx.read();
}
handleLastHttpContent(msg, promise);
return;
}
if (persistentConnection && pendingResponses == 0) {
Expand All @@ -446,6 +426,112 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
ctx.write(msg, promise);
}

@SuppressWarnings("FutureReturnValueIgnored")
boolean handleDefaultFullHttpResponse(DefaultFullHttpResponse response, ChannelPromise promise) {
nonInformationalResponse = !isInformational(response);
// Assume the response writer knows if they can persist or not and sets isKeepAlive on the response
boolean maxKeepAliveRequestsReached = maxKeepAliveRequests != -1 && HttpServerOperations.requestsCounter(ctx.channel()) == maxKeepAliveRequests;
if (maxKeepAliveRequestsReached || !isKeepAlive(response) || !isSelfDefinedMessageLength(response)) {
// No longer keep alive as the client can't tell when the message is done unless we close connection
pendingResponses = 0;
persistentConnection = false;
}
// Server might think it can keep connection alive, but we should fix response header if we know better
if (!shouldKeepAlive()) {
setKeepAlive(response, false);
}

if (response.status().equals(HttpResponseStatus.CONTINUE)) {
//"FutureReturnValueIgnored" this is deliberate
ctx.write(response, promise);
return true;
}
return false;
}

@SuppressWarnings("FutureReturnValueIgnored")
void handleDefaultHttContent(DefaultHttpContent msg, ChannelPromise promise) {
if (persistentConnection && pendingResponses == 0) {
if (HttpServerOperations.log.isDebugEnabled()) {
HttpServerOperations.log.debug(
format(ctx.channel(), "Dropped HTTP content, since response has been sent already: {}"),
httpMessageLogFactory.debug(HttpMessageArgProviderFactory.create(msg)));
}
msg.release();
promise.setSuccess();
return;
}
//"FutureReturnValueIgnored" this is deliberate
ctx.write(msg, promise);
}

@SuppressWarnings("FutureReturnValueIgnored")
void handleDefaultHttpResponse(DefaultHttpResponse response, ChannelPromise promise) {
nonInformationalResponse = !isInformational(response);
// Assume the response writer knows if they can persist or not and sets isKeepAlive on the response
boolean maxKeepAliveRequestsReached = maxKeepAliveRequests != -1 && HttpServerOperations.requestsCounter(ctx.channel()) == maxKeepAliveRequests;
if (maxKeepAliveRequestsReached || !isKeepAlive(response) || !isSelfDefinedMessageLength(response)) {
// No longer keep alive as the client can't tell when the message is done unless we close connection
pendingResponses = 0;
persistentConnection = false;
}
// Server might think it can keep connection alive, but we should fix response header if we know better
if (!shouldKeepAlive()) {
setKeepAlive(response, false);
}
//"FutureReturnValueIgnored" this is deliberate
ctx.write(response, promise);
}

@SuppressWarnings("FutureReturnValueIgnored")
void handleLastHttpContent(Object msg, ChannelPromise promise) {
finalizingResponse = true;

if (LAST_FLUSH_WHEN_NO_READ) {
needsFlush = !read;
}

if (!shouldKeepAlive()) {
if (HttpServerOperations.log.isDebugEnabled()) {
HttpServerOperations.log.debug(format(ctx.channel(), "Detected non persistent http " +
"connection, preparing to close. Pending responses count: {}"),
pendingResponses);
}
//"FutureReturnValueIgnored" this is deliberate
ctx.write(msg, promise.unvoid()).addListener(ChannelFutureListener.CLOSE);
return;
}

//"FutureReturnValueIgnored" this is deliberate
ctx.write(msg, promise);

if (!persistentConnection) {
return;
}

if (nonInformationalResponse) {
nonInformationalResponse = false;
pendingResponses -= 1;
if (HttpServerOperations.log.isDebugEnabled()) {
HttpServerOperations.log.debug(format(ctx.channel(), "Decreasing pending responses count: {}"),
pendingResponses);
}
}

if (pipelined != null && !pipelined.isEmpty()) {
if (HttpServerOperations.log.isDebugEnabled()) {
HttpServerOperations.log.debug(format(ctx.channel(), "Draining next pipelined " +
"HTTP request, pending responses count: {}, queued: {}"),
pendingResponses, pipelined.size());
}
ctx.executor().execute(this);
}
else {
IdleTimeoutHandler.addIdleTimeoutHandler(ctx.pipeline(), idleTimeout);
ctx.read();
}
}

@Override
public void run() {
Object next;
Expand Down