Refactor duplicate code in tests

- Add missing Javadoc
- Do not allocate Message objects needlessly
- Reduce String building
- No need to call constructor with the default value, an AtomicBoolean
is false by default.
- Access ArgumentMatchers static methods directly
- Use try-with-resources

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientH2StreamHandler.java

@@ -110,9 +110,9 @@ public void endStream() throws IOException {
         this.exchangeHandler = exchangeHandler;
         this.pushHandlerFactory = pushHandlerFactory;
         this.context = context;
-        this.requestCommitted = new AtomicBoolean(false);
-        this.failed = new AtomicBoolean(false);
-        this.done = new AtomicBoolean(false);
+        this.requestCommitted = new AtomicBoolean();
+        this.failed = new AtomicBoolean();
+        this.done = new AtomicBoolean();
         this.requestState = MessageState.HEADERS;
         this.responseState = MessageState.HEADERS;
     }

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientPushH2StreamHandler.java

@@ -78,8 +78,8 @@ class ClientPushH2StreamHandler implements H2StreamHandler {
         this.connMetrics = connMetrics;
         this.pushHandlerFactory = pushHandlerFactory;
         this.context = context;
-        this.failed = new AtomicBoolean(false);
-        this.done = new AtomicBoolean(false);
+        this.failed = new AtomicBoolean();
+        this.done = new AtomicBoolean();
         this.requestState = MessageState.HEADERS;
         this.responseState = MessageState.HEADERS;
     }

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ServerH2StreamHandler.java

@@ -137,9 +137,9 @@ public void pushPromise(
         this.connMetrics = connMetrics;
         this.exchangeHandlerFactory = exchangeHandlerFactory;
         this.context = context;
-        this.responseCommitted = new AtomicBoolean(false);
-        this.failed = new AtomicBoolean(false);
-        this.done = new AtomicBoolean(false);
+        this.responseCommitted = new AtomicBoolean();
+        this.failed = new AtomicBoolean();
+        this.done = new AtomicBoolean();
         this.requestState = MessageState.HEADERS;
         this.responseState = MessageState.IDLE;
     }

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ServerPushH2StreamHandler.java

@@ -105,9 +105,9 @@ public void endStream() throws IOException {
         this.connMetrics = connMetrics;
         this.pushProducer = pushProducer;
         this.context = context;
-        this.responseCommitted = new AtomicBoolean(false);
-        this.failed = new AtomicBoolean(false);
-        this.done = new AtomicBoolean(false);
+        this.responseCommitted = new AtomicBoolean();
+        this.failed = new AtomicBoolean();
+        this.done = new AtomicBoolean();
         this.requestState = MessageState.COMPLETE;
         this.responseState = MessageState.IDLE;
     }

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/CancellableExecution.java

@@ -38,7 +38,7 @@ final class CancellableExecution implements CancellableDependency {
     private final AtomicReference<Cancellable> dependencyRef;
 
     CancellableExecution() {
-        this.cancelled = new AtomicBoolean(false);
+        this.cancelled = new AtomicBoolean();
         this.dependencyRef = new AtomicReference<>();
     }
 

httpcore5-h2/src/test/java/org/apache/hc/core5/http2/impl/nio/TestAbstractH2StreamMultiplexer.java

@@ -49,6 +49,7 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentMatchers;
 import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
@@ -139,9 +140,9 @@ void testInputOneFrame() throws Exception {
         Assertions.assertThrows(H2ConnectionException.class, () ->
                 streamMultiplexer.onInput(ByteBuffer.wrap(bytes)));
         Mockito.verify(h2StreamListener).onFrameInput(
-                Mockito.same(streamMultiplexer),
-                Mockito.eq(1),
-                Mockito.any());
+                ArgumentMatchers.same(streamMultiplexer),
+                ArgumentMatchers.eq(1),
+                ArgumentMatchers.any());
 
         Assertions.assertThrows(H2ConnectionException.class, () -> {
             int pos = 0;
@@ -154,9 +155,9 @@ void testInputOneFrame() throws Exception {
             }
 
             Mockito.verify(h2StreamListener).onFrameInput(
-                    Mockito.same(streamMultiplexer),
-                    Mockito.eq(1),
-                    Mockito.any());
+                    ArgumentMatchers.same(streamMultiplexer),
+                    ArgumentMatchers.eq(1),
+                    ArgumentMatchers.any());
         });
     }
 
@@ -190,9 +191,9 @@ void testInputMultipleFrames() throws Exception {
         Assertions.assertThrows(H2ConnectionException.class, () ->
                 streamMultiplexer.onInput(ByteBuffer.wrap(bytes)));
         Mockito.verify(h2StreamListener).onFrameInput(
-                Mockito.same(streamMultiplexer),
-                Mockito.eq(1),
-                Mockito.any());
+                ArgumentMatchers.same(streamMultiplexer),
+                ArgumentMatchers.eq(1),
+                ArgumentMatchers.any());
 
         Assertions.assertThrows(H2ConnectionException.class, () -> {
             int pos = 0;
@@ -205,9 +206,9 @@ void testInputMultipleFrames() throws Exception {
             }
 
             Mockito.verify(h2StreamListener).onFrameInput(
-                    Mockito.same(streamMultiplexer),
-                    Mockito.eq(1),
-                    Mockito.any());
+                    ArgumentMatchers.same(streamMultiplexer),
+                    ArgumentMatchers.eq(1),
+                    ArgumentMatchers.any());
         });
     }
 

httpcore5-reactive/src/main/java/org/apache/hc/core5/reactive/ReactiveDataConsumer.java

@@ -58,7 +58,7 @@ final class ReactiveDataConsumer implements AsyncDataConsumer, Publisher<ByteBuf
     private final AtomicLong requests = new AtomicLong(0);
 
     private final BlockingQueue<ByteBuffer> buffers = new LinkedBlockingQueue<>();
-    private final AtomicBoolean flushInProgress = new AtomicBoolean(false);
+    private final AtomicBoolean flushInProgress = new AtomicBoolean();
     private final AtomicInteger windowScalingIncrement = new AtomicInteger(0);
     private volatile boolean cancelled;
     private volatile boolean completed;

httpcore5-reactive/src/main/java/org/apache/hc/core5/reactive/ReactiveDataProducer.java

@@ -55,7 +55,7 @@ final class ReactiveDataProducer implements AsyncDataProducer, Subscriber<ByteBu
 
     private final AtomicReference<DataStreamChannel> requestChannel = new AtomicReference<>();
     private final AtomicReference<Throwable> exception = new AtomicReference<>();
-    private final AtomicBoolean complete = new AtomicBoolean(false);
+    private final AtomicBoolean complete = new AtomicBoolean();
     private final Publisher<ByteBuffer> publisher;
     private final AtomicReference<Subscription> subscription = new AtomicReference<>();
     private final ArrayDeque<ByteBuffer> buffers = new ArrayDeque<>(); // This field requires synchronization

httpcore5-testing/pom.xml

@@ -70,6 +70,11 @@
       <artifactId>rxjava</artifactId>
       <version>${rxjava3.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.apache.httpcomponents.core5</groupId>
+      <artifactId>httpcore5</artifactId>
+      <classifier>tests</classifier>
+    </dependency>
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
       <artifactId>log4j-slf4j-impl</artifactId>

httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/ClientSessionEndpoint.java

@@ -66,7 +66,7 @@ public final class ClientSessionEndpoint implements ModalCloseable {
     public ClientSessionEndpoint(final IOSession ioSession) {
         super();
         this.ioSession = ioSession;
-        this.closed = new AtomicBoolean(false);
+        this.closed = new AtomicBoolean();
     }
 
     public void execute(final Command command, final Command.Priority priority) {

httpcore5-testing/src/test/java/org/apache/hc/core5/testing/compatibility/http2/H2CompatibilityTest.java

@@ -285,24 +285,16 @@ void executeHttpBin(final HttpHost target) throws Exception {
             System.out.println("*** httpbin.org HTTP/1.1 simple request execution ***");
 
             final List<Message<HttpRequest, AsyncEntityProducer>> requestMessages = Arrays.asList(
-                    new Message<>(
-                            new BasicHttpRequest(Method.GET, target, "/headers"),
-                            null),
+                    new Message<>(new BasicHttpRequest(Method.GET, target, "/headers")),
                     new Message<>(
                             new BasicHttpRequest(Method.POST, target, "/anything"),
                             new StringAsyncEntityProducer("some important message", ContentType.TEXT_PLAIN)),
                     new Message<>(
                             new BasicHttpRequest(Method.PUT, target, "/anything"),
                             new StringAsyncEntityProducer("some important message", ContentType.TEXT_PLAIN)),
-                    new Message<>(
-                            new BasicHttpRequest(Method.GET, target, "/drip"),
-                            null),
-                    new Message<>(
-                            new BasicHttpRequest(Method.GET, target, "/bytes/20000"),
-                            null),
-                    new Message<>(
-                            new BasicHttpRequest(Method.GET, target, "/delay/2"),
-                            null),
+                    new Message<>(new BasicHttpRequest(Method.GET, target, "/drip")),
+                    new Message<>(new BasicHttpRequest(Method.GET, target, "/bytes/20000")),
+                    new Message<>(new BasicHttpRequest(Method.GET, target, "/delay/2")),
                     new Message<>(
                             new BasicHttpRequest(Method.POST, target, "/delay/2"),
                             new StringAsyncEntityProducer("some important message", ContentType.TEXT_PLAIN)),

httpcore5-testing/src/test/java/org/apache/hc/core5/testing/nio/H2ConnPoolTest.java

@@ -28,11 +28,10 @@
 package org.apache.hc.core5.testing.nio;
 
 import java.net.InetSocketAddress;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicLong;
 
-import org.apache.hc.core5.concurrent.FutureCallback;
+import org.apache.hc.core5.concurrent.CountDownLatchFutureCallback;
 import org.apache.hc.core5.function.Supplier;
 import org.apache.hc.core5.http.HttpHost;
 import org.apache.hc.core5.http.URIScheme;
@@ -113,29 +112,18 @@ void testManyGetSession() throws Exception {
 
         final H2MultiplexingRequester requester = clientResource.start();
         final H2ConnPool connPool = requester.getConnPool();
-        final CountDownLatch latch = new CountDownLatch(n);
+        final CountDownLatchFutureCallback<IOSession> latch = new CountDownLatchFutureCallback<IOSession>(n) {
+
+            @Override
+            public void completed(final IOSession session) {
+                session.enqueue(new PingCommand(new BasicPingHandler(
+                        result -> countDown()
+                        )), Command.Priority.IMMEDIATE);
+            }
+
+        };
         for (int i = 0; i < n; i++) {
-            connPool.getSession(target, TIMEOUT, new FutureCallback<IOSession>() {
-
-                @Override
-                public void completed(final IOSession session) {
-                    session.enqueue(new PingCommand(new BasicPingHandler(
-                            result -> {
-                                latch.countDown();
-                            })), Command.Priority.IMMEDIATE);
-                }
-
-                @Override
-                public void failed(final Exception ex) {
-                    latch.countDown();
-                }
-
-                @Override
-                public void cancelled() {
-                    latch.countDown();
-                }
-
-            });
+            connPool.getSession(target, TIMEOUT, latch);
         }
         Assertions.assertTrue(latch.await(TIMEOUT.getDuration(), TIMEOUT.getTimeUnit()));
 
@@ -153,27 +141,10 @@ void testManyGetSessionFailures() throws Exception {
 
         final H2MultiplexingRequester requester = clientResource.start();
         final H2ConnPool connPool = requester.getConnPool();
-        final CountDownLatch latch = new CountDownLatch(n);
+        final CountDownLatchFutureCallback<IOSession> latch = new CountDownLatchFutureCallback<>(n);
 
         for (int i = 0; i < n; i++) {
-            connPool.getSession(target, TIMEOUT, new FutureCallback<IOSession>() {
-
-                @Override
-                public void completed(final IOSession session) {
-                    latch.countDown();
-                }
-
-                @Override
-                public void failed(final Exception ex) {
-                    latch.countDown();
-                }
-
-                @Override
-                public void cancelled() {
-                    latch.countDown();
-                }
-
-            });
+            connPool.getSession(target, TIMEOUT, latch);
         }
 
         requester.initiateShutdown();

httpcore5-testing/src/test/java/org/apache/hc/core5/testing/nio/H2CoreTransportMultiplexingTest.java

@@ -33,11 +33,10 @@
 import java.util.LinkedList;
 import java.util.Queue;
 import java.util.Random;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Future;
 
 import org.apache.hc.core5.concurrent.Cancellable;
-import org.apache.hc.core5.concurrent.FutureCallback;
+import org.apache.hc.core5.concurrent.CountDownLatchFutureCallback;
 import org.apache.hc.core5.function.Supplier;
 import org.apache.hc.core5.http.ContentType;
 import org.apache.hc.core5.http.HttpHost;
@@ -233,32 +232,15 @@ void testMultiplexedRequestCancellation() throws Exception {
 
         final int reqNo = 20;
 
-        final CountDownLatch countDownLatch = new CountDownLatch(reqNo);
+        final CountDownLatchFutureCallback<Message<HttpResponse, String>> countDownLatch = new CountDownLatchFutureCallback<>(reqNo);
         final Random random = new Random();
         final HttpHost target = new HttpHost(scheme.id, "localhost", address.getPort());
         for (int i = 0; i < reqNo; i++) {
             final Cancellable cancellable = requester.execute(
                     new BasicClientExchangeHandler<>(new BasicRequestProducer(Method.POST, target, "/stuff",
                             new StringAsyncEntityProducer("some stuff", ContentType.TEXT_PLAIN)),
                             new BasicResponseConsumer<>(new StringAsyncEntityConsumer()),
-                            new FutureCallback<Message<HttpResponse, String>>() {
-
-                                @Override
-                                public void completed(final Message<HttpResponse, String> result) {
-                                    countDownLatch.countDown();
-                                }
-
-                                @Override
-                                public void failed(final Exception ex) {
-                                    countDownLatch.countDown();
-                                }
-
-                                @Override
-                                public void cancelled() {
-                                    countDownLatch.countDown();
-                                }
-
-                            }),
+                            countDownLatch),
                     TIMEOUT,
                     HttpCoreContext.create());
             Thread.sleep(random.nextInt(10));

httpcore5-testing/src/test/java/org/apache/hc/core5/testing/reactive/ReactiveClientTest.java

@@ -211,7 +211,7 @@ void testRequestError() throws Exception {
     void testRequestTimeout() throws Exception {
         final InetSocketAddress address = startServer();
         final HttpAsyncRequester requester = clientResource.start();
-        final AtomicBoolean requestPublisherWasCancelled = new AtomicBoolean(false);
+        final AtomicBoolean requestPublisherWasCancelled = new AtomicBoolean();
         final Publisher<ByteBuffer> publisher = Flowable.<ByteBuffer>never()
                 .doOnCancel(() -> requestPublisherWasCancelled.set(true));
         final ReactiveEntityProducer producer = new ReactiveEntityProducer(publisher, -1, null, null);
@@ -237,7 +237,7 @@ void testRequestTimeout() throws Exception {
     void testResponseCancellation() throws Exception {
         final InetSocketAddress address = startServer();
         final HttpAsyncRequester requester = clientResource.start();
-        final AtomicBoolean requestPublisherWasCancelled = new AtomicBoolean(false);
+        final AtomicBoolean requestPublisherWasCancelled = new AtomicBoolean();
         final AtomicReference<Throwable> requestStreamError = new AtomicReference<>();
         final Publisher<ByteBuffer> stream = Reactive3TestUtils.produceStream(Long.MAX_VALUE, 1024, null)
                 .doOnCancel(() -> requestPublisherWasCancelled.set(true))
@@ -250,7 +250,7 @@ void testResponseCancellation() throws Exception {
         final Message<HttpResponse, Publisher<ByteBuffer>> response = consumer.getResponseFuture()
                 .get(RESULT_TIMEOUT.getDuration(), RESULT_TIMEOUT.getTimeUnit());
 
-        final AtomicBoolean responsePublisherWasCancelled = new AtomicBoolean(false);
+        final AtomicBoolean responsePublisherWasCancelled = new AtomicBoolean();
         final List<ByteBuffer> outputBuffers = Flowable.fromPublisher(response.getBody())
                 .doOnCancel(() -> responsePublisherWasCancelled.set(true))
                 .take(3)

httpcore5/src/main/java/org/apache/hc/core5/concurrent/FutureCallback.java

@@ -29,7 +29,7 @@
 /**
  * A callback interface that gets invoked upon completion of a {@link java.util.concurrent.Future}.
  *
- * @param <T> the future result type returned by this callback.
+ * @param <T> the future result type consumed by this callback.
  * @since 4.2
  */
 public interface FutureCallback<T> {

httpcore5/src/main/java/org/apache/hc/core5/concurrent/FutureContribution.java

@@ -37,6 +37,11 @@ public abstract class FutureContribution<T> implements FutureCallback<T> {
 
     private final BasicFuture<?> future;
 
+    /**
+     * Constructs a new instance to callback the given {@link BasicFuture}.
+     *
+     * @param future The callback.
+     */
     public FutureContribution(final BasicFuture<?> future) {
         this.future = future;
     }