CXF-8911: Allow wrapping AsyncHTTPConduit response processing (using new AsyncHttpResponseWrapperFactory bus extension)

rt/transports/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduit.java

@@ -66,6 +66,7 @@
 import org.apache.cxf.transport.http.Headers;
 import org.apache.cxf.transport.http.HttpClientHTTPConduit;
 import org.apache.cxf.transport.http.asyncclient.AsyncHTTPConduitFactory.UseAsyncPolicy;
+import org.apache.cxf.transport.http.asyncclient.AsyncHttpResponseWrapperFactory.AsyncHttpResponseWrapper;
 import org.apache.cxf.transport.https.HttpsURLConnectionInfo;
 import org.apache.cxf.transports.http.configuration.HTTPClientPolicy;
 import org.apache.cxf.version.Version;
@@ -99,13 +100,15 @@
 public class AsyncHTTPConduit extends HttpClientHTTPConduit {
     public static final String USE_ASYNC = "use.async.http.conduit";
 
-    final AsyncHTTPConduitFactory factory;
-    volatile int lastTlsHash = -1;
-    volatile Object sslState;
-    volatile URI sslURL;
-    volatile SSLContext sslContext;
-    volatile SSLSession session;
-    volatile CloseableHttpAsyncClient client;
+    private final AsyncHTTPConduitFactory factory;
+    private final AsyncHttpResponseWrapperFactory asyncHttpResponseWrapperFactory;
+
+    private volatile int lastTlsHash = -1;
+    private volatile Object sslState;
+    private volatile URI sslURL;
+    private volatile SSLContext sslContext;
+    private volatile SSLSession session;
+    private volatile CloseableHttpAsyncClient client;
 
 
     public AsyncHTTPConduit(Bus b,
@@ -114,6 +117,7 @@ public AsyncHTTPConduit(Bus b,
                             AsyncHTTPConduitFactory factory) throws IOException {
         super(b, ei, t);
         this.factory = factory;
+        this.asyncHttpResponseWrapperFactory = bus.getExtension(AsyncHttpResponseWrapperFactory.class);
     }
 
     public synchronized CloseableHttpAsyncClient getHttpAsyncClient() throws IOException {
@@ -478,14 +482,26 @@ protected void connect(boolean output) throws IOException {
                 return;
             }
 
-            CXFResponseCallback responseCallback = new CXFResponseCallback() {
+            CXFResponseCallback delegate = new CXFResponseCallback() {
                 @Override
                 public void responseReceived(HttpResponse response) {
                     setHttpResponse(response);
                 }
 
             };
 
+            CXFResponseCallback responseCallback = delegate;
+            if (asyncHttpResponseWrapperFactory != null) {
+                final AsyncHttpResponseWrapper wrapper = asyncHttpResponseWrapperFactory.create();
+                if (wrapper != null) {
+                    responseCallback = new CXFResponseCallback() {
+                        @Override
+                        public void responseReceived(HttpResponse response) {
+                            wrapper.responseReceived(response, delegate::responseReceived);
+                        }
+                    };
+                }
+            }
             FutureCallback<Boolean> callback = new FutureCallback<Boolean>() {
 
                 public void completed(Boolean result) {

rt/transports/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/AsyncHttpResponseWrapperFactory.java

@@ -0,0 +1,55 @@
+/**
+ * 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.
+ */
+
+package org.apache.cxf.transport.http.asyncclient;
+
+import java.util.function.Consumer;
+
+import org.apache.cxf.Bus;
+import org.apache.http.HttpResponse;
+
+/**
+ * The {@link Bus} extension to allow wrapping up the response processing of 
+ * the {@link AsyncHTTPConduit} instance.
+ */
+@FunctionalInterface
+public interface AsyncHttpResponseWrapperFactory {
+    /** 
+     * Creates new instance of the {@link AsyncHttpResponseWrapper} 
+     * @return new instance of the {@link AsyncHttpResponseWrapper} (or null)
+     */
+    AsyncHttpResponseWrapper create();
+
+    /**
+     * The wrapper around the response that will be called by the {@link AsyncHTTPConduit} 
+     * instance once the response is received. 
+     */
+    interface AsyncHttpResponseWrapper {
+        /**
+         * The callback which is called by the {@link AsyncHTTPConduit} instance once 
+         * the response is received. The delegating response handler is passed as the
+         * an argument and has to be called.
+         * @param response the response received
+         * @param delegate delegating response handler
+         */
+        default void responseReceived(HttpResponse response, Consumer<HttpResponse> delegate) {
+            delegate.accept(response);
+        }
+    }
+}

rt/transports/http-hc/src/test/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduitTest.java

@@ -26,6 +26,7 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
 
 import jakarta.xml.ws.AsyncHandler;
 import jakarta.xml.ws.Endpoint;
@@ -43,19 +44,23 @@
 import org.apache.cxf.testutil.common.AbstractBusClientServerTestBase;
 import org.apache.cxf.transport.http.HTTPConduit;
 import org.apache.cxf.transport.http.HTTPConduitFactory;
+import org.apache.cxf.transport.http.asyncclient.AsyncHttpResponseWrapperFactory.AsyncHttpResponseWrapper;
 import org.apache.cxf.transports.http.configuration.HTTPClientPolicy;
 import org.apache.cxf.workqueue.AutomaticWorkQueueImpl;
 import org.apache.cxf.workqueue.WorkQueueManager;
 import org.apache.hello_world_soap_http.Greeter;
 import org.apache.hello_world_soap_http.SOAPService;
 import org.apache.hello_world_soap_http.types.GreetMeLaterResponse;
 import org.apache.hello_world_soap_http.types.GreetMeResponse;
+import org.apache.http.HttpResponse;
 
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Ignore;
 import org.junit.Test;
 
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.fail;
@@ -242,7 +247,7 @@ public void testConnectIssue() throws Exception {
     }
 
     @Test
-    public void testInovationWithHCAddress() throws Exception {
+    public void testInvocationWithHCAddress() throws Exception {
         String address = "hc://http://localhost:" + PORT + "/SoapContext/SoapPort";
         JaxWsProxyFactoryBean factory = new JaxWsProxyFactoryBean();
         factory.setServiceClass(Greeter.class);
@@ -263,6 +268,7 @@ public void testInvocationWithTransportId() throws Exception {
         String response = greeter.greetMe("test");
         assertEquals("Get a wrong response", "Hello test", response);
     }
+
     @Test
     public void testCall() throws Exception {
         updateAddressPort(g, PORT);
@@ -273,6 +279,7 @@ public void testCall() throws Exception {
         c.setClient(cp);
         assertEquals("Hello " + request, g.greetMe(request));
     }
+
     @Test
     public void testCallAsync() throws Exception {
         updateAddressPort(g, PORT);
@@ -293,6 +300,35 @@ public void handleResponse(Response<GreetMeLaterResponse> res) {
         }).get();
     }
 
+    @Test
+    public void testCallAsyncWithResponseWrapper() throws Exception {
+        try {
+            final CountDownLatch latch = new CountDownLatch(1);
+            final AsyncHttpResponseWrapper wrapper = new AsyncHttpResponseWrapper() {
+                @Override
+                public void responseReceived(HttpResponse response, Consumer<HttpResponse> delegate) {
+                    delegate.accept(response);
+                    latch.countDown();
+                }
+            };
+
+            getStaticBus().setExtension(() -> wrapper, AsyncHttpResponseWrapperFactory.class);
+
+            final String address = "hc://http://localhost:" + PORT + "/SoapContext/SoapPort";
+            final Greeter greeter = new SOAPService().getSoapPort();
+            setAddress(greeter, address);
+
+            greeter.greetMeLaterAsync(1000, new AsyncHandler<GreetMeLaterResponse>() {
+                public void handleResponse(Response<GreetMeLaterResponse> res) {
+                }
+            }).get();
+
+            assertThat(latch.await(5, TimeUnit.SECONDS), is(true));
+        } finally {
+            getStaticBus().setExtension(null, AsyncHttpResponseWrapperFactory.class);
+        }
+    }
+
     @Test
     public void testCallAsyncCallbackInvokedOnlyOnce() throws Exception {
         // This test is especially targeted for RHEL 6.8

rt/transports/http-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/AsyncHTTPConduit.java

@@ -66,6 +66,7 @@
 import org.apache.cxf.transport.http.Headers;
 import org.apache.cxf.transport.http.HttpClientHTTPConduit;
 import org.apache.cxf.transport.http.asyncclient.hc5.AsyncHTTPConduitFactory.UseAsyncPolicy;
+import org.apache.cxf.transport.http.asyncclient.hc5.AsyncHttpResponseWrapperFactory.AsyncHttpResponseWrapper;
 import org.apache.cxf.transport.https.HttpsURLConnectionInfo;
 import org.apache.cxf.transports.http.configuration.HTTPClientPolicy;
 import org.apache.cxf.version.Version;
@@ -105,6 +106,7 @@ public class AsyncHTTPConduit extends HttpClientHTTPConduit {
     public static final String USE_ASYNC = "use.async.http.conduit";
 
     private final AsyncHTTPConduitFactory factory;
+    private final AsyncHttpResponseWrapperFactory asyncHttpResponseWrapperFactory;
     private volatile int lastTlsHash = -1;
     private volatile Object sslState;
     private volatile URI sslURL;
@@ -116,6 +118,7 @@ public AsyncHTTPConduit(Bus b, EndpointInfo ei, EndpointReferenceType t, AsyncHT
             throws IOException {
         super(b, ei, t);
         this.factory = factory;
+        this.asyncHttpResponseWrapperFactory = bus.getExtension(AsyncHttpResponseWrapperFactory.class);
     }
 
     public synchronized CloseableHttpAsyncClient getHttpAsyncClient(final TlsStrategy tlsStrategy)
@@ -491,15 +494,29 @@ protected void connect(boolean output) throws IOException {
             if (connectionFuture != null) {
                 return;
             }
-
-            CXFResponseCallback responseCallback = new CXFResponseCallback() {
+            
+            final CXFResponseCallback delegate = new CXFResponseCallback() {
                 @Override
                 public void responseReceived(HttpResponse response) {
                     setHttpResponse(response);
                 }
 
             };
 
+            CXFResponseCallback responseCallback = delegate;
+            if (asyncHttpResponseWrapperFactory != null) {
+                final AsyncHttpResponseWrapper wrapper = asyncHttpResponseWrapperFactory.create();
+                if (wrapper != null) {
+                    responseCallback = new CXFResponseCallback() {
+                        @Override
+                        public void responseReceived(HttpResponse response) {
+                            wrapper.responseReceived(response, delegate::responseReceived);
+                        }
+                    };
+                }
+            }
+
+
             FutureCallback<Boolean> callback = new FutureCallback<Boolean>() {
 
                 public void completed(Boolean result) {

rt/transports/http-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/AsyncHttpResponseWrapperFactory.java

@@ -0,0 +1,55 @@
+/**
+ * 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.
+ */
+
+package org.apache.cxf.transport.http.asyncclient.hc5;
+
+import java.util.function.Consumer;
+
+import org.apache.cxf.Bus;
+import org.apache.hc.core5.http.HttpResponse;
+
+/**
+ * The {@link Bus} extension to allow wrapping up the response processing of 
+ * the {@link AsyncHTTPConduit} instance.
+ */
+@FunctionalInterface
+public interface AsyncHttpResponseWrapperFactory {
+    /** 
+     * Creates new instance of the {@link AsyncHttpResponseWrapper} 
+     * @return new instance of the {@link AsyncHttpResponseWrapper} (or null)
+     */
+    AsyncHttpResponseWrapper create();
+
+    /**
+     * The wrapper around the response that will be called by the {@link AsyncHTTPConduit} 
+     * instance once the response is received. 
+     */
+    interface AsyncHttpResponseWrapper {
+        /**
+         * The callback which is called by the {@link AsyncHTTPConduit} instance once 
+         * the response is received. The delegating response handler is passed as the
+         * an argument and has to be called.
+         * @param response the response received
+         * @param delegate delegating response handler
+         */
+        default void responseReceived(HttpResponse response, Consumer<HttpResponse> delegate) {
+            delegate.accept(response);
+        }
+    }
+}