diff --git a/jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannel.java b/jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannel.java index 1f787a0356c9..9134588cfd99 100644 --- a/jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannel.java +++ b/jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannel.java @@ -1195,10 +1195,7 @@ public void succeeded() if (_length > 0) _combinedListener.onResponseContent(_request, _content); if (_complete && _state.completeResponse()) - { - _response.getHttpOutput().closed(); _combinedListener.onResponseEnd(_request); - } super.succeeded(); } @@ -1222,7 +1219,6 @@ public void succeeded() @Override public void failed(Throwable th) { - _response.getHttpOutput().closed(); abort(x); super.failed(x); } diff --git a/jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannelState.java b/jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannelState.java index 53e302d63113..bf3027591200 100644 --- a/jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannelState.java +++ b/jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannelState.java @@ -968,6 +968,9 @@ protected void completed() } } + // release any aggregate buffer from a closing flush + _channel.getResponse().getHttpOutput().closed(); + if (event != null) { cancelTimeout(event); diff --git a/jetty-server/src/main/java/org/eclipse/jetty/server/HttpOutput.java b/jetty-server/src/main/java/org/eclipse/jetty/server/HttpOutput.java index 70fbf4868e74..f651179862ab 100644 --- a/jetty-server/src/main/java/org/eclipse/jetty/server/HttpOutput.java +++ b/jetty-server/src/main/java/org/eclipse/jetty/server/HttpOutput.java @@ -234,6 +234,8 @@ private void write(ByteBuffer content, boolean complete) throws IOException { write(content, complete, blocker); blocker.block(); + if (complete) + closed(); } catch (Exception failure) { @@ -403,13 +405,6 @@ public void closed() State state = _state.get(); switch (state) { - case CLOSING: - { - if (!_state.compareAndSet(state, State.CLOSED)) - break; - releaseBuffer(); - return; - } case CLOSED: { return; @@ -417,7 +412,12 @@ public void closed() case UNREADY: { if (_state.compareAndSet(state, State.ERROR)) - _writeListener.onError(_onError == null ? new EofException("Async closed") : _onError); + { + if (_onError == null) + _onError = new EofException("Async closed"); + releaseBuffer(); + return; + } break; } default: diff --git a/jetty-server/src/test/java/org/eclipse/jetty/server/AsyncCompletionTest.java b/jetty-server/src/test/java/org/eclipse/jetty/server/AsyncCompletionTest.java index a14a6bcd09f4..376b9f33baa0 100644 --- a/jetty-server/src/test/java/org/eclipse/jetty/server/AsyncCompletionTest.java +++ b/jetty-server/src/test/java/org/eclipse/jetty/server/AsyncCompletionTest.java @@ -33,6 +33,12 @@ import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Stream; +import javax.servlet.AsyncContext; +import javax.servlet.ServletException; +import javax.servlet.ServletOutputStream; +import javax.servlet.WriteListener; +import javax.servlet.http.HttpServletRequest; +import javax.servlet.http.HttpServletResponse; import org.eclipse.jetty.http.HttpCompliance; import org.eclipse.jetty.http.HttpTester; @@ -41,6 +47,7 @@ import org.eclipse.jetty.io.EndPoint; import org.eclipse.jetty.io.ManagedSelector; import org.eclipse.jetty.io.SocketChannelEndPoint; +import org.eclipse.jetty.server.handler.AbstractHandler; import org.eclipse.jetty.util.Callback; import org.eclipse.jetty.util.thread.Scheduler; import org.hamcrest.Matchers; @@ -89,7 +96,7 @@ public void proceed() _delay.get(10, TimeUnit.SECONDS); getCallback().succeeded(); } - catch(Throwable th) + catch (Throwable th) { th.printStackTrace(); getCallback().failed(th); @@ -97,7 +104,6 @@ public void proceed() } } - @BeforeEach public void init() throws Exception { @@ -153,7 +159,7 @@ public ExtendedHttpConnection(HttpConfiguration config, Connector connector, End @Override public void onCompleted() { - COMPLETE.compareAndSet(false,true); + COMPLETE.compareAndSet(false, true); super.onCompleted(); } } @@ -163,7 +169,8 @@ public static Stream tests() { List tests = new ArrayList<>(); tests.add(new Object[]{new HelloWorldHandler(), 200, "Hello world"}); - tests.add(new Object[]{new SendErrorHandler(499,"Test async sendError"), 499, "Test async sendError"}); + tests.add(new Object[]{new SendErrorHandler(499, "Test async sendError"), 499, "Test async sendError"}); + tests.add(new Object[]{new AsyncReadyCompleteHandler(), 200, AsyncReadyCompleteHandler.data}); return tests.stream().map(Arguments::of); } @@ -197,7 +204,7 @@ public void testAsyncCompletion(Handler handler, int status, String message) thr // wait for threads to return to base level long end = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); - while(_threadPool.getBusyThreads() != base) + while (_threadPool.getBusyThreads() != base) { if (System.nanoTime() > end) throw new TimeoutException(); @@ -210,7 +217,7 @@ public void testAsyncCompletion(Handler handler, int status, String message) thr // proceed with the completion delay.proceed(); - while(!COMPLETE.get()) + while (!COMPLETE.get()) { if (System.nanoTime() > end) throw new TimeoutException(); @@ -218,4 +225,46 @@ public void testAsyncCompletion(Handler handler, int status, String message) thr } } } + + private static class AsyncReadyCompleteHandler extends AbstractHandler + { + static String data = "Now is the time for all good men to come to the aid of the party"; + + @Override + public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException + { + AsyncContext context = request.startAsync(); + ServletOutputStream out = response.getOutputStream(); + out.setWriteListener(new WriteListener() + { + byte[] bytes = data.getBytes(StandardCharsets.ISO_8859_1); + + @Override + public void onWritePossible() throws IOException + { + while (out.isReady()) + { + if (bytes != null) + { + response.setContentType("text/plain"); + response.setContentLength(bytes.length); + out.write(bytes); + bytes = null; + } + else + { + context.complete(); + return; + } + } + } + + @Override + public void onError(Throwable t) + { + t.printStackTrace(); + } + }); + } + } }