Fixed the problem with request specific push handlers being ignored; …

…added push response integration test

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

@@ -0,0 +1,80 @@
+/*
+ * ====================================================================
+ * 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;
+
+import org.apache.hc.core5.http.HttpRequest;
+import org.apache.hc.core5.http.HttpResponse;
+import org.apache.hc.core5.http.message.RequestLine;
+import org.apache.hc.core5.http.message.StatusLine;
+
+public final class Result<T> {
+
+    public final HttpRequest request;
+    public final HttpResponse response;
+    public final T content;
+    public final Exception exception;
+
+    public enum Status { OK, NOK }
+
+    public Result(final HttpRequest request, final Exception exception) {
+        this.request = request;
+        this.response = null;
+        this.content = null;
+        this.exception = exception;
+    }
+
+    public Result(final HttpRequest request, final HttpResponse response, final T content) {
+        this.request = request;
+        this.response = response;
+        this.content = content;
+        this.exception = null;
+    }
+
+    public Status getStatus() {
+        return exception != null ? Status.NOK : Status.OK;
+    }
+
+    public boolean isOK() {
+        return exception == null;
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder buf = new StringBuilder();
+        buf.append(new RequestLine(request));
+        buf.append(" -> ");
+        if (exception != null) {
+            buf.append("NOK: ").append(exception);
+        } else {
+            if (response != null) {
+                buf.append("OK: ").append(new StatusLine(response));
+            }
+        }
+        return buf.toString();
+    }
+
+}

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

@@ -0,0 +1,206 @@
+/*
+ * ====================================================================
+ * 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 java.io.IOException;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
+
+import org.apache.hc.client5.http.async.methods.SimpleHttpRequest;
+import org.apache.hc.client5.http.async.methods.SimpleHttpResponse;
+import org.apache.hc.client5.http.async.methods.SimpleRequestBuilder;
+import org.apache.hc.client5.http.async.methods.SimpleRequestProducer;
+import org.apache.hc.client5.http.async.methods.SimpleResponseConsumer;
+import org.apache.hc.client5.http.protocol.HttpClientContext;
+import org.apache.hc.client5.testing.Result;
+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.concurrent.FutureCallback;
+import org.apache.hc.core5.http.ContentType;
+import org.apache.hc.core5.http.EntityDetails;
+import org.apache.hc.core5.http.HttpException;
+import org.apache.hc.core5.http.HttpHost;
+import org.apache.hc.core5.http.HttpRequest;
+import org.apache.hc.core5.http.HttpStatus;
+import org.apache.hc.core5.http.RequestNotExecutedException;
+import org.apache.hc.core5.http.URIScheme;
+import org.apache.hc.core5.http.nio.AsyncRequestConsumer;
+import org.apache.hc.core5.http.nio.AsyncServerRequestHandler;
+import org.apache.hc.core5.http.nio.entity.DiscardingEntityConsumer;
+import org.apache.hc.core5.http.nio.entity.StringAsyncEntityProducer;
+import org.apache.hc.core5.http.nio.support.AbstractAsyncPushHandler;
+import org.apache.hc.core5.http.nio.support.AbstractAsyncRequesterConsumer;
+import org.apache.hc.core5.http.nio.support.AbstractServerExchangeHandler;
+import org.apache.hc.core5.http.nio.support.BasicPushProducer;
+import org.apache.hc.core5.http.nio.support.BasicResponseProducer;
+import org.apache.hc.core5.http.protocol.HttpContext;
+import org.apache.hc.core5.http.support.BasicRequestBuilder;
+import org.apache.hc.core5.http.support.BasicResponseBuilder;
+import org.apache.hc.core5.http2.config.H2Config;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+abstract class AbstractH2AsyncFundamentalsTest extends AbstractHttpAsyncFundamentalsTest {
+
+    public AbstractH2AsyncFundamentalsTest(final URIScheme scheme, final ClientProtocolLevel clientProtocolLevel, final ServerProtocolLevel serverProtocolLevel) {
+        super(scheme, clientProtocolLevel, serverProtocolLevel);
+    }
+
+    @Test
+    void testPush() throws Exception {
+        configureServer(bootstrap -> bootstrap
+                .register("/pushy", () -> new AbstractServerExchangeHandler<HttpRequest>() {
+
+                    @Override
+                    protected AsyncRequestConsumer<HttpRequest> supplyConsumer(
+                            final HttpRequest request,
+                            final EntityDetails entityDetails,
+                            final HttpContext context) throws HttpException {
+
+                        return new AbstractAsyncRequesterConsumer<HttpRequest, Void>(new DiscardingEntityConsumer<>()) {
+
+                            @Override
+                            protected HttpRequest buildResult(final HttpRequest request, final Void entity, final ContentType contentType) {
+                                return request;
+                            }
+
+                        };
+                    }
+
+                    @Override
+                    protected void handle(
+                            final HttpRequest request,
+                            final AsyncServerRequestHandler.ResponseTrigger responseTrigger,
+                            final HttpContext context) throws HttpException, IOException {
+                        responseTrigger.pushPromise(
+                                BasicRequestBuilder.copy(request)
+                                        .setPath("/aaa")
+                                        .build(),
+                                context,
+                                new BasicPushProducer(BasicResponseBuilder.create(HttpStatus.SC_OK)
+                                        .build(),
+                                        new StringAsyncEntityProducer("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", ContentType.TEXT_PLAIN)));
+                        responseTrigger.pushPromise(
+                                BasicRequestBuilder.copy(request)
+                                        .setPath("/bbb")
+                                        .build(),
+                                context,
+                                new BasicPushProducer(
+                                        BasicResponseBuilder.create(HttpStatus.SC_OK).build(),
+                                        new StringAsyncEntityProducer("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", ContentType.TEXT_PLAIN)));
+                        responseTrigger.submitResponse(
+                                new BasicResponseProducer(
+                                        BasicResponseBuilder.create(HttpStatus.SC_OK).build(),
+                                        new StringAsyncEntityProducer("I am being very pushy")
+                                ),
+                                context);
+                    }
+
+                }));
+
+        configureClient(builder -> builder
+                .setH2Config(H2Config.custom()
+                        .setPushEnabled(true)
+                        .build()));
+
+        final HttpHost target = startServer();
+
+        final TestAsyncClient client = startClient();
+
+        client.start();
+
+        final Queue<Result<String>> pushMessageQueue = new ConcurrentLinkedQueue<>();
+        final CountDownLatch latch = new CountDownLatch(3);
+        final HttpClientContext context = HttpClientContext.create();
+        final SimpleHttpRequest request = SimpleRequestBuilder.get()
+                .setHttpHost(target)
+                .setPath("/pushy")
+                .build();
+        client.execute(
+                SimpleRequestProducer.create(request),
+                SimpleResponseConsumer.create(),
+                (r, c) -> new AbstractAsyncPushHandler<SimpleHttpResponse>(SimpleResponseConsumer.create()) {
+
+                    @Override
+                    protected void handleResponse(final HttpRequest promise,
+                                                  final SimpleHttpResponse response) throws IOException, HttpException {
+                        pushMessageQueue.add(new Result<>(promise, response, response.getBodyText()));
+                        latch.countDown();
+                    }
+
+                    @Override
+                    protected void handleError(final HttpRequest promise, final Exception cause) {
+                        pushMessageQueue.add(new Result<>(promise, cause));
+                        latch.countDown();
+                    }
+
+                },
+                context,
+                new FutureCallback<SimpleHttpResponse>() {
+
+                    @Override
+                    public void completed(final SimpleHttpResponse response) {
+                        pushMessageQueue.add(new Result<>(request, response, response.getBodyText()));
+                        latch.countDown();
+                    }
+
+                    @Override
+                    public void failed(final Exception ex) {
+                        pushMessageQueue.add(new Result<>(request, ex));
+                        latch.countDown();
+                    }
+
+                    @Override
+                    public void cancelled() {
+                        pushMessageQueue.add(new Result<>(request, new RequestNotExecutedException()));
+                        latch.countDown();
+                    }
+
+                }
+        );
+        Assertions.assertTrue(latch.await(TIMEOUT.getDuration(), TIMEOUT.getTimeUnit()));
+        Assertions.assertEquals(3, pushMessageQueue.size());
+        for (final Result<String> result : pushMessageQueue) {
+            if (result.isOK()) {
+                Assertions.assertEquals(HttpStatus.SC_OK, result.response.getCode());
+                final String path = result.request.getPath();
+                if (path.equals("/pushy")) {
+                    Assertions.assertEquals("I am being very pushy", result.content);
+                } else if (path.equals("/aaa")) {
+                    Assertions.assertEquals("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", result.content);
+                } else if (path.equals("/bbb")) {
+                    Assertions.assertEquals("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", result.content);
+                } else {
+                    Assertions.fail("Unxpected request path: " + path);
+                }
+            }
+        }
+    }
+
+}

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

@@ -30,7 +30,7 @@
 import org.apache.hc.client5.testing.extension.async.ServerProtocolLevel;
 import org.apache.hc.core5.http.URIScheme;
 
-abstract  class TestH2Async extends AbstractHttpAsyncFundamentalsTest {
+abstract class TestH2Async extends AbstractH2AsyncFundamentalsTest {
 
     public TestH2Async(final URIScheme scheme) {
         super(scheme, ClientProtocolLevel.H2_ONLY, ServerProtocolLevel.H2_ONLY);

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

@@ -30,7 +30,7 @@
 import org.apache.hc.client5.testing.extension.async.ServerProtocolLevel;
 import org.apache.hc.core5.http.URIScheme;
 
-abstract  class TestH2AsyncMinimal extends AbstractHttpAsyncFundamentalsTest {
+abstract class TestH2AsyncMinimal extends AbstractH2AsyncFundamentalsTest {
 
     public TestH2AsyncMinimal(final URIScheme scheme) {
         super(scheme, ClientProtocolLevel.MINIMAL_H2_ONLY, ServerProtocolLevel.H2_ONLY);

httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/InternalHttpAsyncExecRuntime.java

@@ -286,7 +286,7 @@ public Cancellable execute(
             if (responseTimeout != null) {
                 endpoint.setSocketTimeout(responseTimeout);
             }
-            endpoint.execute(id, exchangeHandler, context);
+            endpoint.execute(id, exchangeHandler, pushHandlerFactory, context);
             if (context.getRequestConfigOrDefault().isHardCancellationEnabled()) {
                 return () -> {
                     exchangeHandler.cancel();