diff --git a/Jenkinsfile b/Jenkinsfile index 4e193c586b3..18886fbe488 100644 --- a/Jenkinsfile +++ b/Jenkinsfile @@ -39,6 +39,17 @@ pipeline { } } + stage("Build / Test - JDK23") { + agent { node { label 'linux' } } + steps { + timeout( time: 180, unit: 'MINUTES' ) { + checkout scm + mavenBuild( "jdk23", "clean install -Dspotbugs.skip=true -Djacoco.skip=true", "maven3") + recordIssues id: "jdk23", name: "Static Analysis jdk23", aggregatingResults: true, enabledForFailure: true, tools: [mavenConsole(), java(), checkStyle(), javaDoc()] + } + } + } + stage("Build / Test - JDK17") { agent { node { label 'linux' } } steps { diff --git a/jetty-core/jetty-http/src/main/java/org/eclipse/jetty/http/HttpFields.java b/jetty-core/jetty-http/src/main/java/org/eclipse/jetty/http/HttpFields.java index 05bed910083..66db59325e3 100644 --- a/jetty-core/jetty-http/src/main/java/org/eclipse/jetty/http/HttpFields.java +++ b/jetty-core/jetty-http/src/main/java/org/eclipse/jetty/http/HttpFields.java @@ -699,7 +699,7 @@ public interface HttpFields extends Iterable, Supplier * @return the value of the field as a {@code long}, * or -1 if no such field is present * @throws NumberFormatException if the value of the field - * cannot be converted to a {@link long} + * cannot be converted to a {@code long} */ default long getLongField(String name) throws NumberFormatException { @@ -715,7 +715,7 @@ public interface HttpFields extends Iterable, Supplier * @return the value of the field as a {@code long}, * or -1 if no such field is present * @throws NumberFormatException if the value of the field - * cannot be converted to a {@link long} + * cannot be converted to a {@code long} */ default long getLongField(HttpHeader header) throws NumberFormatException { diff --git a/jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/internal/HttpChannelState.java b/jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/internal/HttpChannelState.java index bd9a75581c6..92f9bcf43df 100644 --- a/jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/internal/HttpChannelState.java +++ b/jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/internal/HttpChannelState.java @@ -482,7 +482,8 @@ public class HttpChannelState implements HttpChannel, Components } } - // Consume content as soon as possible to open any flow control window. + // Consume content as soon as possible to open any + // flow control window and release any request buffer. Throwable unconsumed = stream.consumeAvailable(); if (unconsumed != null && LOG.isDebugEnabled()) LOG.debug("consuming content during error {}", unconsumed.toString()); diff --git a/jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/internal/HttpConnection.java b/jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/internal/HttpConnection.java index ebb030467bb..6905a3ce258 100644 --- a/jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/internal/HttpConnection.java +++ b/jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/internal/HttpConnection.java @@ -19,6 +19,7 @@ import java.nio.channels.WritePendingException; import java.util.List; import java.util.Objects; import java.util.Set; +import java.util.concurrent.Executor; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; @@ -103,7 +104,7 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab private final LongAdder bytesOut = new LongAdder(); private final AtomicBoolean _handling = new AtomicBoolean(false); private final HttpFields.Mutable _headerBuilder = HttpFields.build(); - private volatile RetainableByteBuffer _retainableByteBuffer; + private volatile RetainableByteBuffer _requestBuffer; private HttpFields.Mutable _trailers; private Runnable _onRequest; private long _requests; @@ -319,21 +320,18 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab public ByteBuffer onUpgradeFrom() { if (isRequestBufferEmpty()) - { - releaseRequestBuffer(); return null; - } - ByteBuffer unconsumed = ByteBuffer.allocateDirect(_retainableByteBuffer.remaining()); - unconsumed.put(_retainableByteBuffer.getByteBuffer()); + ByteBuffer unconsumed = ByteBuffer.allocateDirect(_requestBuffer.remaining()); + unconsumed.put(_requestBuffer.getByteBuffer()); unconsumed.flip(); - releaseRequestBuffer(); return unconsumed; } @Override public void onUpgradeTo(ByteBuffer buffer) { - BufferUtil.append(getRequestBuffer(), buffer); + ensureRequestBuffer(); + BufferUtil.append(_requestBuffer.getByteBuffer(), buffer); } @Override @@ -345,51 +343,49 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab void releaseRequestBuffer() { - if (_retainableByteBuffer != null && _retainableByteBuffer.isEmpty()) - { - if (LOG.isDebugEnabled()) - LOG.debug("releaseRequestBuffer {}", this); - RetainableByteBuffer buffer = _retainableByteBuffer; - _retainableByteBuffer = null; - if (!buffer.release()) - throw new IllegalStateException("unreleased buffer " + buffer); - } + if (LOG.isDebugEnabled()) + LOG.debug("releasing request buffer {} {}", _requestBuffer, this); + _requestBuffer.release(); + _requestBuffer = null; } - private ByteBuffer getRequestBuffer() + private void ensureRequestBuffer() { - if (_retainableByteBuffer == null) - _retainableByteBuffer = _bufferPool.acquire(getInputBufferSize(), isUseInputDirectByteBuffers()); - return _retainableByteBuffer.getByteBuffer(); + if (_requestBuffer == null) + { + _requestBuffer = _bufferPool.acquire(getInputBufferSize(), isUseInputDirectByteBuffers()); + if (LOG.isDebugEnabled()) + LOG.debug("request buffer acquired {} {}", _requestBuffer, this); + } } public boolean isRequestBufferEmpty() { - return _retainableByteBuffer == null || _retainableByteBuffer.isEmpty(); + return _requestBuffer == null || !_requestBuffer.hasRemaining(); } @Override public void onFillable() { if (LOG.isDebugEnabled()) - LOG.debug(">>onFillable enter {} {} {}", this, _httpChannel, _retainableByteBuffer); + LOG.debug("onFillable enter {} {} {}", _httpChannel, _requestBuffer, this); HttpConnection last = setCurrentConnection(this); try { + ensureRequestBuffer(); + // We must loop until we fill -1 or there is an async pause in handling. // Note that the endpoint might already be closed in some special circumstances. while (true) { - // Fill the request buffer (if needed). int filled = fillRequestBuffer(); if (LOG.isDebugEnabled()) - LOG.debug("onFillable filled {} {} {} {}", filled, this, _httpChannel, _retainableByteBuffer); + LOG.debug("onFillable filled {} {} {} {}", filled, _httpChannel, _requestBuffer, this); if (filled < 0 && getEndPoint().isOutputShutdown()) close(); - // Parse the request buffer. boolean handle = parseRequestBuffer(); // There could be a connection upgrade before handling @@ -397,52 +393,66 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab // If there was a connection upgrade, the other // connection took over, nothing more to do here. if (getEndPoint().getConnection() != this) + { + releaseRequestBuffer(); break; + } - // Handle channel event. This will only be true when the headers of a request have been received. + // The headers of a request have been received. if (handle) { Request request = _httpChannel.getRequest(); if (LOG.isDebugEnabled()) LOG.debug("HANDLE {} {}", request, this); - // handle the request by running the task obtained from onRequest + // Handle the request by running the task. _handling.set(true); Runnable onRequest = _onRequest; _onRequest = null; onRequest.run(); - // If the _handling boolean has already been CaS'd to false, then stream is completed and we are no longer - // handling, so the caller can continue to fill and parse more connections. If it is still true, then some - // thread is still handling the request and they will need to organize more filling and parsing once complete. + // If the CaS succeeds, then some thread is still handling the request. + // If the CaS fails, then stream is completed, we are no longer handling, + // so the caller can continue to fill and parse more connections. if (_handling.compareAndSet(true, false)) { if (LOG.isDebugEnabled()) - LOG.debug("request !complete {} {}", request, this); + LOG.debug("request !complete {} {} {}", request, _requestBuffer, this); + // Cannot release the request buffer here, because the + // application may read concurrently from another thread. + // The request buffer will be released by the application + // reading the request content, or by the implementation + // trying to consume the request content. break; } - // If the request is complete, but has been upgraded, then break + // If there was an upgrade, release and return. if (getEndPoint().getConnection() != this) { if (LOG.isDebugEnabled()) LOG.debug("upgraded {} -> {}", this, getEndPoint().getConnection()); + releaseRequestBuffer(); break; } } + else if (filled == 0) + { + assert isRequestBufferEmpty(); + releaseRequestBuffer(); + fillInterested(); + break; + } else if (filled < 0) { + assert isRequestBufferEmpty(); + releaseRequestBuffer(); getEndPoint().shutdownOutput(); break; } else if (_requestHandler._failure != null) { // There was an error, don't fill more. - break; - } - else if (filled == 0) - { - fillInterested(); + releaseRequestBuffer(); break; } } @@ -453,11 +463,8 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab { if (LOG.isDebugEnabled()) LOG.debug("caught exception {} {}", this, _httpChannel, x); - if (_retainableByteBuffer != null) - { - _retainableByteBuffer.clear(); + if (_requestBuffer != null) releaseRequestBuffer(); - } } finally { @@ -468,7 +475,7 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab { setCurrentConnection(last); if (LOG.isDebugEnabled()) - LOG.debug("< 0) - { bytesIn.add(filled); - } - else - { - if (filled < 0) - _parser.atEOF(); - releaseRequestBuffer(); - } + else if (filled < 0) + _parser.atEOF(); return filled; } @@ -550,11 +550,6 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab if (LOG.isDebugEnabled()) LOG.debug("Unable to fill from endpoint {}", getEndPoint(), x); _parser.atEOF(); - if (_retainableByteBuffer != null) - { - _retainableByteBuffer.clear(); - releaseRequestBuffer(); - } return -1; } } @@ -562,19 +557,15 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab private boolean parseRequestBuffer() { if (LOG.isDebugEnabled()) - LOG.debug("{} parse {}", this, _retainableByteBuffer); + LOG.debug("parse {} {}", _requestBuffer, this); if (_parser.isTerminated()) throw new RuntimeIOException("Parser is terminated"); - boolean handle = _parser.parseNext(_retainableByteBuffer == null ? BufferUtil.EMPTY_BUFFER : _retainableByteBuffer.getByteBuffer()); + boolean handle = _parser.parseNext(_requestBuffer.getByteBuffer()); if (LOG.isDebugEnabled()) - LOG.debug("{} parsed {} {}", this, handle, _parser); - - // recycle buffer ? - if (_retainableByteBuffer != null && !_retainableByteBuffer.isRetained()) - releaseRequestBuffer(); + LOG.debug("parsed {} {} {} {}", handle, _parser, _requestBuffer, this); return handle; } @@ -631,19 +622,6 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab getExecutor().execute(this); } - @Override - public void onClose(Throwable cause) - { - // TODO: do we really need to do this? - // This event is fired really late, sendCallback should already be failed at this point. - // Revisit whether we still need IteratingCallback.close(). - if (cause == null) - _sendCallback.close(); - else - _sendCallback.abort(cause); - super.onClose(cause); - } - @Override public void run() { @@ -989,14 +967,14 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab public boolean content(ByteBuffer buffer) { HttpStreamOverHTTP1 stream = _stream.get(); - if (stream == null || stream._chunk != null || _retainableByteBuffer == null) + if (stream == null || stream._chunk != null || _requestBuffer == null) throw new IllegalStateException(); if (LOG.isDebugEnabled()) - LOG.debug("content {}/{} for {}", BufferUtil.toDetailString(buffer), _retainableByteBuffer, HttpConnection.this); + LOG.debug("content {}/{} for {}", BufferUtil.toDetailString(buffer), _requestBuffer, HttpConnection.this); - _retainableByteBuffer.retain(); - stream._chunk = Content.Chunk.asChunk(buffer, false, _retainableByteBuffer); + _requestBuffer.retain(); + stream._chunk = Content.Chunk.asChunk(buffer, false, _requestBuffer); return true; } @@ -1540,6 +1518,11 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab if (LOG.isDebugEnabled()) LOG.debug("abort due to pending read {} {} ", this, getEndPoint()); abort(new IOException("Pending read in onCompleted")); + _httpChannel.recycle(); + _parser.reset(); + _generator.reset(); + if (!_handling.compareAndSet(true, false)) + resume(); return; } @@ -1549,6 +1532,8 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab _httpChannel.recycle(); _parser.close(); _generator.reset(); + if (!_handling.compareAndSet(true, false)) + releaseRequestBuffer(); return; } @@ -1580,40 +1565,7 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab if (LOG.isDebugEnabled()) LOG.debug("non-current completion {}", this); - // If we are looking for the next request - if (_parser.isStart()) - { - // if the buffer is empty - if (isRequestBufferEmpty()) - { - // look for more data - fillInterested(); - } - // else if we are still running - else if (getConnector().isRunning()) - { - // Dispatched to handle a pipelined request - try - { - getExecutor().execute(HttpConnection.this); - } - catch (RejectedExecutionException e) - { - if (getConnector().isRunning()) - LOG.warn("Failed dispatch of {}", this, e); - else - LOG.trace("IGNORED", e); - getEndPoint().close(); - } - } - else - { - getEndPoint().close(); - } - } - // else the parser must be closed, so seek the EOF if we are still open - else if (getEndPoint().isOpen()) - fillInterested(); + resume(); } @Override @@ -1629,6 +1581,30 @@ public class HttpConnection extends AbstractMetaDataConnection implements Runnab if (LOG.isDebugEnabled()) LOG.debug("aborting", x); abort(x); + _httpChannel.recycle(); + _parser.reset(); + _generator.reset(); + if (!_handling.compareAndSet(true, false)) + resume(); + } + + private void resume() + { + try + { + if (LOG.isDebugEnabled()) + LOG.debug("Resuming onFillable() {}", HttpConnection.this); + // Dispatch to handle pipelined requests. + Request request = _httpChannel.getRequest(); + Executor executor = request == null ? getExecutor() : request.getComponents().getExecutor(); + executor.execute(HttpConnection.this); + } + catch (RejectedExecutionException x) + { + getEndPoint().close(x); + // Resume by running, to release the request buffer. + run(); + } } private void abort(Throwable failure) diff --git a/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/HttpServerTestBase.java b/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/HttpServerTestBase.java index f86408c292a..6cf5573bc50 100644 --- a/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/HttpServerTestBase.java +++ b/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/HttpServerTestBase.java @@ -60,7 +60,7 @@ import org.hamcrest.Matchers; import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.ValueSource; +import org.junit.jupiter.params.provider.CsvSource; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -68,7 +68,6 @@ import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.endsWith; import static org.hamcrest.Matchers.equalTo; -import static org.hamcrest.Matchers.greaterThan; import static org.hamcrest.Matchers.greaterThanOrEqualTo; import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.is; @@ -1869,8 +1868,8 @@ public abstract class HttpServerTestBase extends HttpServerTestFixture } @ParameterizedTest - @ValueSource(booleans = {false /* TODO, true */}) - public void testHoldContent(boolean close) throws Exception + @CsvSource({"false,false", "false,true", "true,false", "true,true"}) + public void testHoldContent(boolean close, boolean pipeline) throws Exception { Queue contents = new ConcurrentLinkedQueue<>(); final int bufferSize = 1024; @@ -1881,6 +1880,15 @@ public abstract class HttpServerTestBase extends HttpServerTestFixture @Override public boolean handle(Request request, Response response, Callback callback) throws Exception { + if (LOG.isDebugEnabled()) + LOG.debug("Handling request: {}", request); + if ("GET".equals(request.getMethod())) + { + response.setStatus(200); + callback.succeeded(); + return true; + } + request.getConnectionMetaData().getConnection().addEventListener(new Connection.Listener() { @Override @@ -1889,10 +1897,12 @@ public abstract class HttpServerTestBase extends HttpServerTestFixture closed.countDown(); } }); + while (true) { Content.Chunk chunk = request.read(); - + if (LOG.isDebugEnabled()) + LOG.debug("read: {}", chunk); if (chunk == null) { try (Blocker.Runnable blocker = Blocker.runnable()) @@ -1902,81 +1912,111 @@ public abstract class HttpServerTestBase extends HttpServerTestFixture continue; } } - if (chunk.hasRemaining()) contents.add(chunk); else chunk.release(); - if (chunk.isLast()) break; } - response.setStatus(200); if (close) + { + LOG.info("Closing {}", request.getConnectionMetaData().getConnection().getEndPoint()); request.getConnectionMetaData().getConnection().getEndPoint().close(); + } callback.succeeded(); return true; } }); - byte[] chunk = new byte[bufferSize / 2]; Arrays.fill(chunk, (byte)'X'); - try (Socket client = newSocket(_serverURI.getHost(), _serverURI.getPort())) { OutputStream os = client.getOutputStream(); BufferedOutputStream out = new BufferedOutputStream(os, bufferSize); - out.write((""" + String request = """ POST / HTTP/1.1\r Host: localhost\r - Connection: close\r + Connection: %s\r Transfer-Encoding: chunked\r \r - """).getBytes(StandardCharsets.ISO_8859_1)); + """.formatted(pipeline ? "other" : "close"); + if (LOG.isDebugEnabled()) + LOG.debug("raw request {}", request); + out.write(request.getBytes(StandardCharsets.ISO_8859_1)); // single chunk out.write((Integer.toHexString(chunk.length) + "\r\n").getBytes(StandardCharsets.ISO_8859_1)); out.write(chunk); out.write("\r\n".getBytes(StandardCharsets.ISO_8859_1)); out.flush(); - - // double chunk (will overflow) + // double chunk (will overflow bufferSize) out.write((Integer.toHexString(chunk.length * 2) + "\r\n").getBytes(StandardCharsets.ISO_8859_1)); out.write(chunk); out.write(chunk); out.write("\r\n".getBytes(StandardCharsets.ISO_8859_1)); out.flush(); - // single chunk and end chunk - out.write((Integer.toHexString(chunk.length) + "\r\n").getBytes(StandardCharsets.ISO_8859_1)); - out.write(chunk); - out.write("\r\n0\r\n\r\n".getBytes(StandardCharsets.ISO_8859_1)); + // single chunk and end chunk plus optional pipelined request + ByteBuffer last = BufferUtil.allocate(4096); + BufferUtil.append(last, (Integer.toHexString(chunk.length) + "\r\n").getBytes(StandardCharsets.ISO_8859_1)); + BufferUtil.append(last, chunk); + BufferUtil.append(last, "\r\n0\r\n\r\n".getBytes(StandardCharsets.ISO_8859_1)); + if (pipeline) + BufferUtil.append(last, "GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n".getBytes(StandardCharsets.ISO_8859_1)); + + if (LOG.isDebugEnabled()) + LOG.debug("last {}", BufferUtil.toString(last)); + out.write(BufferUtil.toArray(last)); out.flush(); // check the response - if (!close) + if (close) + { + assertThat(client.getInputStream().read(), equalTo(-1)); + } + else { HttpTester.Response response = HttpTester.parseResponse(client.getInputStream()); assertNotNull(response); assertThat(response.getStatus(), is(200)); + + if (pipeline) + { + response = HttpTester.parseResponse(client.getInputStream()); + assertNotNull(response); + assertThat(response.getStatus(), is(200)); + } } } assertTrue(closed.await(10, TimeUnit.SECONDS)); - long total = contents.stream().mapToLong(Content.Chunk::remaining).sum(); assertThat(total, equalTo(chunk.length * 4L)); - ByteBufferPool rbbp = _connector.getByteBufferPool(); if (rbbp instanceof ArrayByteBufferPool pool) { long buffersBeforeRelease = pool.getAvailableDirectByteBufferCount() + pool.getAvailableHeapByteBufferCount(); + if (LOG.isDebugEnabled()) + { + LOG.debug("pool {}", pool); + contents.stream().map(Content.Chunk::toString).forEach(LOG::debug); + } contents.forEach(Content.Chunk::release); - long buffersAfterRelease = pool.getAvailableDirectByteBufferCount() + pool.getAvailableHeapByteBufferCount(); - assertThat(buffersAfterRelease, greaterThan(buffersBeforeRelease)); + + Awaitility.waitAtMost(5, TimeUnit.SECONDS).until(() -> + { + if (LOG.isDebugEnabled()) + { + LOG.debug("pool {}", pool); + contents.stream().map(Content.Chunk::toString).forEach(LOG::debug); + } + long buffersAfterRelease = pool.getAvailableDirectByteBufferCount() + pool.getAvailableHeapByteBufferCount(); + return buffersAfterRelease > buffersBeforeRelease; + }); assertThat(pool.getAvailableDirectMemory() + pool.getAvailableHeapMemory(), greaterThanOrEqualTo(chunk.length * 4L)); } else diff --git a/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/RequestListenersTest.java b/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/RequestListenersTest.java index 9282fa8f256..ba6d61eac85 100644 --- a/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/RequestListenersTest.java +++ b/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/RequestListenersTest.java @@ -191,11 +191,13 @@ public class RequestListenersTest Content-Length: 1 Connection: close - """, 2000 * idleTimeout, TimeUnit.MILLISECONDS)); + """, 2 * idleTimeout, TimeUnit.MILLISECONDS)); int expectedStatus = succeedCallback ? HttpStatus.OK_200 : HttpStatus.INTERNAL_SERVER_ERROR_500; assertEquals(expectedStatus, response.getStatus()); - assertThat(failureLatch.await(idleTimeout + 500, TimeUnit.MILLISECONDS), is(failIdleTimeout && !succeedCallback)); + // The failure listener is never invoked because completing the callback + // produces a response that completes the stream so the failure is ignored. + assertThat(failureLatch.await(idleTimeout + 500, TimeUnit.MILLISECONDS), is(false)); } @ParameterizedTest diff --git a/jetty-core/jetty-tests/jetty-test-client-transports/src/test/java/org/eclipse/jetty/test/client/transport/HttpClientIdleTimeoutTest.java b/jetty-core/jetty-tests/jetty-test-client-transports/src/test/java/org/eclipse/jetty/test/client/transport/HttpClientIdleTimeoutTest.java index 9a969015e03..01cdf57c672 100644 --- a/jetty-core/jetty-tests/jetty-test-client-transports/src/test/java/org/eclipse/jetty/test/client/transport/HttpClientIdleTimeoutTest.java +++ b/jetty-core/jetty-tests/jetty-test-client-transports/src/test/java/org/eclipse/jetty/test/client/transport/HttpClientIdleTimeoutTest.java @@ -15,6 +15,8 @@ package org.eclipse.jetty.test.client.transport; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicReference; import org.eclipse.jetty.client.ContentResponse; import org.eclipse.jetty.client.StringRequestContent; @@ -26,6 +28,8 @@ import org.eclipse.jetty.util.Callback; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.MethodSource; +import static org.awaitility.Awaitility.await; +import static org.hamcrest.Matchers.notNullValue; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -38,17 +42,17 @@ public class HttpClientIdleTimeoutTest extends AbstractTest public void testClientIdleTimeout(Transport transport) throws Exception { long serverIdleTimeout = idleTimeout * 2; - CountDownLatch serverIdleTimeoutLatch = new CountDownLatch(1); + AtomicReference serverCallbackRef = new AtomicReference<>(); start(transport, new Handler.Abstract() { @Override public boolean handle(Request request, Response response, Callback callback) { // Do not succeed the callback if it's a timeout request. - if (!Request.getPathInContext(request).equals("/timeout")) - callback.succeeded(); + if (Request.getPathInContext(request).equals("/timeout")) + request.addFailureListener(x -> serverCallbackRef.set(callback)); else - request.addFailureListener(x -> serverIdleTimeoutLatch.countDown()); + callback.succeeded(); return true; } }); @@ -75,7 +79,8 @@ public class HttpClientIdleTimeoutTest extends AbstractTest assertEquals(HttpStatus.OK_200, response.getStatus()); // Wait for the server's idle timeout to trigger to give it a chance to clean up its resources. - assertTrue(serverIdleTimeoutLatch.await(2 * serverIdleTimeout, TimeUnit.MILLISECONDS)); + Callback callback = await().atMost(2 * serverIdleTimeout, TimeUnit.MILLISECONDS).until(serverCallbackRef::get, notNullValue()); + callback.failed(new TimeoutException()); } @ParameterizedTest @@ -83,17 +88,17 @@ public class HttpClientIdleTimeoutTest extends AbstractTest public void testRequestIdleTimeout(Transport transport) throws Exception { long serverIdleTimeout = idleTimeout * 2; - CountDownLatch serverIdleTimeoutLatch = new CountDownLatch(1); + AtomicReference serverCallbackRef = new AtomicReference<>(); start(transport, new Handler.Abstract() { @Override public boolean handle(Request request, Response response, Callback callback) throws Exception { // Do not succeed the callback if it's a timeout request. - if (!Request.getPathInContext(request).equals("/timeout")) - callback.succeeded(); + if (Request.getPathInContext(request).equals("/timeout")) + request.addFailureListener(x -> serverCallbackRef.set(callback)); else - request.addFailureListener(x -> serverIdleTimeoutLatch.countDown()); + callback.succeeded(); return true; } }); @@ -120,7 +125,8 @@ public class HttpClientIdleTimeoutTest extends AbstractTest assertEquals(HttpStatus.OK_200, response.getStatus()); // Wait for the server's idle timeout to trigger to give it a chance to clean up its resources. - assertTrue(serverIdleTimeoutLatch.await(2 * serverIdleTimeout, TimeUnit.MILLISECONDS)); + Callback callback = await().atMost(2 * serverIdleTimeout, TimeUnit.MILLISECONDS).until(serverCallbackRef::get, notNullValue()); + callback.failed(new TimeoutException()); } @ParameterizedTest