Message multiplexing for the standard async HttpClient

httpclient5-testing/src/test/java/org/apache/hc/client5/testing/async/HttpIntegrationTests.java

@@ -213,4 +213,24 @@ public AuthenticationH2Tls() throws Exception {
 
     }
 
+    @Nested
+    @DisplayName("HTTP message multiplexing (HTTP/2)")
+    class RequestMultiplexing extends TestHttpAsyncRequestMultiplexing {
+
+        public RequestMultiplexing() {
+            super(URIScheme.HTTP);
+        }
+
+    }
+
+    @Nested
+    @DisplayName("HTTP message multiplexing (HTTP/2, TLS)")
+    class RequestMultiplexingTls extends TestHttpAsyncRequestMultiplexing {
+
+        public RequestMultiplexingTls() {
+            super(URIScheme.HTTPS);
+        }
+
+    }
+
 }

httpclient5-testing/src/test/java/org/apache/hc/client5/testing/async/TestHttpAsyncRequestMultiplexing.java

@@ -0,0 +1,98 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation.  For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+package org.apache.hc.client5.testing.async;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+
+import java.util.LinkedList;
+import java.util.Queue;
+import java.util.Random;
+import java.util.concurrent.Future;
+
+import org.apache.hc.client5.http.config.TlsConfig;
+import org.apache.hc.client5.http.protocol.HttpClientContext;
+import org.apache.hc.client5.testing.extension.async.ClientProtocolLevel;
+import org.apache.hc.client5.testing.extension.async.ServerProtocolLevel;
+import org.apache.hc.client5.testing.extension.async.TestAsyncClient;
+import org.apache.hc.core5.http.ContentType;
+import org.apache.hc.core5.http.HttpHost;
+import org.apache.hc.core5.http.HttpResponse;
+import org.apache.hc.core5.http.Message;
+import org.apache.hc.core5.http.Method;
+import org.apache.hc.core5.http.URIScheme;
+import org.apache.hc.core5.http.nio.entity.AsyncEntityProducers;
+import org.apache.hc.core5.http.nio.entity.BasicAsyncEntityConsumer;
+import org.apache.hc.core5.http.nio.support.BasicRequestProducer;
+import org.apache.hc.core5.http.nio.support.BasicResponseConsumer;
+import org.apache.hc.core5.http2.HttpVersionPolicy;
+import org.hamcrest.CoreMatchers;
+import org.junit.jupiter.api.Test;
+
+abstract class TestHttpAsyncRequestMultiplexing extends AbstractIntegrationTestBase {
+
+    public TestHttpAsyncRequestMultiplexing(final URIScheme uriScheme) {
+        super(uriScheme, ClientProtocolLevel.MINIMAL, ServerProtocolLevel.H2_ONLY);
+    }
+
+    @Test
+    void testConcurrentPostRequests() throws Exception {
+        configureServer(bootstrap -> bootstrap.register("/echo/*", AsyncEchoHandler::new));
+        configureClient(custimizer -> custimizer
+                .setDefaultTlsConfig(TlsConfig.custom()
+                        .setVersionPolicy(HttpVersionPolicy.FORCE_HTTP_2)
+                        .build())
+                .useMessageMultiplexing()
+        );
+        final HttpHost target = startServer();
+        final TestAsyncClient client = startClient();
+        final byte[] b1 = new byte[1024];
+        final Random rnd = new Random(System.currentTimeMillis());
+        rnd.nextBytes(b1);
+
+        final int reqCount = 200;
+
+        final Queue<Future<Message<HttpResponse, byte[]>>> queue = new LinkedList<>();
+        for (int i = 0; i < reqCount; i++) {
+            final Future<Message<HttpResponse, byte[]>> future = client.execute(
+                    new BasicRequestProducer(Method.POST, target, "/echo/",
+                            AsyncEntityProducers.create(b1, ContentType.APPLICATION_OCTET_STREAM)),
+                    new BasicResponseConsumer<>(new BasicAsyncEntityConsumer()), HttpClientContext.create(), null);
+            queue.add(future);
+        }
+
+        while (!queue.isEmpty()) {
+            final Future<Message<HttpResponse, byte[]>> future = queue.remove();
+            final Message<HttpResponse, byte[]> responseMessage = future.get();
+            assertThat(responseMessage, CoreMatchers.notNullValue());
+            final HttpResponse response = responseMessage.getHead();
+            assertThat(response.getCode(), CoreMatchers.equalTo(200));
+            final byte[] b2 = responseMessage.getBody();
+            assertThat(b1, CoreMatchers.equalTo(b2));
+        }
+    }
+
+}

httpclient5-testing/src/test/java/org/apache/hc/client5/testing/compatibility/async/HttpAsyncClientCompatibilityTest.java

@@ -26,6 +26,9 @@
  */
 package org.apache.hc.client5.testing.compatibility.async;
 
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Future;
 
 import org.apache.hc.client5.http.ContextBuilder;
@@ -38,10 +41,13 @@
 import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
 import org.apache.hc.client5.http.impl.auth.BasicCredentialsProvider;
 import org.apache.hc.client5.http.protocol.HttpClientContext;
+import org.apache.hc.client5.testing.Result;
 import org.apache.hc.client5.testing.extension.async.HttpAsyncClientResource;
+import org.apache.hc.core5.concurrent.FutureCallback;
 import org.apache.hc.core5.http.HttpHost;
 import org.apache.hc.core5.http.HttpStatus;
 import org.apache.hc.core5.http.HttpVersion;
+import org.apache.hc.core5.http.RequestNotExecutedException;
 import org.apache.hc.core5.http2.HttpVersionPolicy;
 import org.apache.hc.core5.util.Timeout;
 import org.junit.jupiter.api.Assertions;
@@ -51,6 +57,7 @@
 public abstract class HttpAsyncClientCompatibilityTest {
 
     static final Timeout TIMEOUT = Timeout.ofSeconds(5);
+    static final Timeout LONG_TIMEOUT = Timeout.ofSeconds(30);
 
     private final HttpVersionPolicy versionPolicy;
     private final HttpHost target;
@@ -119,6 +126,54 @@ void test_sequential_gets() throws Exception {
         }
     }
 
+    @Test
+    void test_concurrent_gets() throws Exception {
+        final CloseableHttpAsyncClient client = client();
+
+        final String[] requestUris = new String[] {"/111", "/222", "/333"};
+        final int n = 200;
+        final Queue<Result<Void>> queue = new ConcurrentLinkedQueue<>();
+        final CountDownLatch latch = new CountDownLatch(requestUris.length * n);
+
+        for (int i = 0; i < n; i++) {
+            for (final String requestUri: requestUris) {
+                final SimpleHttpRequest request = SimpleRequestBuilder.get()
+                        .setHttpHost(target)
+                        .setPath(requestUri)
+                        .build();
+                final HttpClientContext context = context();
+                client.execute(request, context, new FutureCallback<SimpleHttpResponse>() {
+
+                    @Override
+                    public void completed(final SimpleHttpResponse response) {
+                        queue.add(new Result<>(request, response, null));
+                        latch.countDown();
+                    }
+
+                    @Override
+                    public void failed(final Exception ex) {
+                        queue.add(new Result<>(request, ex));
+                        latch.countDown();
+                    }
+
+                    @Override
+                    public void cancelled() {
+                        queue.add(new Result<>(request, new RequestNotExecutedException()));
+                        latch.countDown();
+                    }
+
+                });
+            }
+        }
+        Assertions.assertTrue(latch.await(LONG_TIMEOUT.getDuration(), LONG_TIMEOUT.getTimeUnit()));
+        Assertions.assertEquals(requestUris.length * n, queue.size());
+        for (final Result<Void> result : queue) {
+            if (result.isOK()) {
+                Assertions.assertEquals(HttpStatus.SC_OK, result.response.getCode());
+            }
+        }
+    }
+
     @Test
     void test_auth_failure_wrong_auth_scope() throws Exception {
         addCredentials(

httpclient5-testing/src/test/java/org/apache/hc/client5/testing/extension/async/H2OnlyMinimalTestClientBuilder.java

@@ -68,6 +68,11 @@ public TestAsyncClientBuilder setH2Config(final H2Config h2Config) {
         return this;
     }
 
+    @Override
+    public TestAsyncClientBuilder useMessageMultiplexing() {
+        return this;
+    }
+
     @Override
     public TestAsyncClient build() throws Exception {
         final CloseableHttpAsyncClient client = HttpAsyncClients.createHttp2Minimal(

httpclient5-testing/src/test/java/org/apache/hc/client5/testing/extension/async/H2OnlyTestClientBuilder.java

@@ -97,6 +97,11 @@ public TestAsyncClientBuilder setTlsStrategy(final TlsStrategy tlsStrategy) {
         return this;
     }
 
+    @Override
+    public TestAsyncClientBuilder useMessageMultiplexing() {
+        return this;
+    }
+
     @Override
     public TestAsyncClientBuilder setH2Config(final H2Config h2Config) {
         this.h2Config = h2Config;

httpclient5-testing/src/test/java/org/apache/hc/client5/testing/extension/async/HttpAsyncClientResource.java

@@ -61,6 +61,7 @@ public HttpAsyncClientResource(final HttpVersionPolicy versionPolicy) throws IOE
                             .setDefaultTlsConfig(TlsConfig.custom()
                                     .setVersionPolicy(versionPolicy)
                                     .build())
+                            .setMessageMultiplexing(true)
                             .build());
         } catch (final CertificateException | NoSuchAlgorithmException | KeyStoreException | KeyManagementException ex) {
             throw new IllegalStateException(ex);

httpclient5-testing/src/test/java/org/apache/hc/client5/testing/extension/async/MinimalTestClientBuilder.java

@@ -77,6 +77,12 @@ public TestAsyncClientBuilder setDefaultTlsConfig(final TlsConfig tlsConfig) {
         return this;
     }
 
+    @Override
+    public TestAsyncClientBuilder useMessageMultiplexing() {
+        this.connectionManagerBuilder.setMessageMultiplexing(true);
+        return this;
+    }
+
     @Override
     public TestAsyncClientBuilder setHttp1Config(final Http1Config http1Config) {
         this.http1Config = http1Config;

httpclient5-testing/src/test/java/org/apache/hc/client5/testing/extension/async/StandardTestClientBuilder.java

@@ -111,6 +111,12 @@ public TestAsyncClientBuilder setDefaultTlsConfig(final TlsConfig tlsConfig) {
         return this;
     }
 
+    @Override
+    public TestAsyncClientBuilder useMessageMultiplexing() {
+        this.connectionManagerBuilder.setMessageMultiplexing(true);
+        return this;
+    }
+
     @Override
     public TestAsyncClientBuilder setHttp1Config(final Http1Config http1Config) {
         this.clientBuilder.setHttp1Config(http1Config);

httpclient5-testing/src/test/java/org/apache/hc/client5/testing/extension/async/TestAsyncClientBuilder.java

@@ -73,6 +73,10 @@ default TestAsyncClientBuilder setDefaultTlsConfig(TlsConfig tlsConfig) {
         throw new UnsupportedOperationException("Operation not supported by " + getProtocolLevel());
     }
 
+    default TestAsyncClientBuilder useMessageMultiplexing() {
+        throw new UnsupportedOperationException("Operation not supported by " + getProtocolLevel());
+    }
+
     default TestAsyncClientBuilder setHttp1Config(Http1Config http1Config) {
         throw new UnsupportedOperationException("Operation not supported by " + getProtocolLevel());
     }