Redesign of classic over async compatibility layer

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

@@ -1544,6 +1544,14 @@ boolean localReset(final H2Error error) throws IOException {
             return localReset(error != null ? error.getCode() : H2Error.INTERNAL_ERROR.getCode());
         }
 
+        @Override
+        public void terminate() {
+            try {
+                localReset(H2Error.INTERNAL_ERROR);
+            } catch (final IOException ignore) {
+            }
+        }
+
         @Override
         public boolean cancel() {
             try {

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

@@ -43,4 +43,6 @@ interface H2StreamChannel extends DataStreamChannel, CapacityChannel, Cancellabl
 
     void push(List<Header> headers, AsyncPushProducer pushProducer) throws HttpException, IOException;
 
+    void terminate();
+
 }

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

@@ -133,6 +133,11 @@ public void pushPromise(
                 commitPromise(promise, pushProducer);
             }
 
+            @Override
+            public void terminateExchange() {
+                terminate();
+            }
+
         };
         this.httpProcessor = httpProcessor;
         this.connMetrics = connMetrics;
@@ -206,6 +211,10 @@ private void commitPromise(
         connMetrics.incrementRequestCount();
     }
 
+    private void terminate() {
+        outputChannel.terminate();
+    }
+
     @Override
     public void consumePromise(final List<Header> headers) throws HttpException, IOException {
         throw new ProtocolException("Unexpected message promise");

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

@@ -195,6 +195,10 @@ private void commitPromise(
         connMetrics.incrementRequestCount();
     }
 
+    private void terminate() {
+        outputChannel.terminate();
+    }
+
     @Override
     public void produceOutput() throws HttpException, IOException {
         switch (responseState) {
@@ -219,6 +223,11 @@ public void pushPromise(
                         commitPromise(promise, pushProducer);
                     }
 
+                    @Override
+                    public void terminateExchange() {
+                        terminate();
+                    }
+
                 }, context);
                 break;
             case BODY:

httpcore5-h2/src/test/java/org/apache/hc/core5/http2/examples/ClassicH2RequestExecutionExample.java

@@ -0,0 +1,155 @@
+/*
+ * ====================================================================
+ * 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.core5.http2.examples;
+
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.concurrent.Future;
+
+import org.apache.hc.core5.annotation.Experimental;
+import org.apache.hc.core5.http.ClassicHttpRequest;
+import org.apache.hc.core5.http.ClassicHttpResponse;
+import org.apache.hc.core5.http.ContentType;
+import org.apache.hc.core5.http.Header;
+import org.apache.hc.core5.http.HttpConnection;
+import org.apache.hc.core5.http.HttpEntity;
+import org.apache.hc.core5.http.HttpHost;
+import org.apache.hc.core5.http.impl.bootstrap.HttpAsyncRequester;
+import org.apache.hc.core5.http.io.support.ClassicRequestBuilder;
+import org.apache.hc.core5.http.nio.AsyncClientEndpoint;
+import org.apache.hc.core5.http.nio.support.classic.ClassicToAsyncRequestProducer;
+import org.apache.hc.core5.http.nio.support.classic.ClassicToAsyncResponseConsumer;
+import org.apache.hc.core5.http2.HttpVersionPolicy;
+import org.apache.hc.core5.http2.config.H2Config;
+import org.apache.hc.core5.http2.frame.RawFrame;
+import org.apache.hc.core5.http2.impl.nio.H2StreamListener;
+import org.apache.hc.core5.http2.impl.nio.bootstrap.H2RequesterBootstrap;
+import org.apache.hc.core5.io.CloseMode;
+import org.apache.hc.core5.util.Timeout;
+
+/**
+ * Example of HTTP/2 request execution with a classic I/O API compatibility bridge
+ * that enables the use of standard {@link java.io.InputStream} / {@link java.io.OutputStream}
+ * based data consumers / producers.
+ * <p>>
+ * Execution of individual message exchanges is performed at the current thread.
+ */
+@Experimental
+public class ClassicH2RequestExecutionExample {
+
+    public static void main(final String[] args) throws Exception {
+
+        // Create and start requester
+        final H2Config h2Config = H2Config.custom()
+                .setPushEnabled(false)
+                .build();
+
+        final HttpAsyncRequester requester = H2RequesterBootstrap.bootstrap()
+                .setH2Config(h2Config)
+                .setVersionPolicy(HttpVersionPolicy.FORCE_HTTP_2)
+                .setStreamListener(new H2StreamListener() {
+
+                    @Override
+                    public void onHeaderInput(final HttpConnection connection, final int streamId, final List<? extends Header> headers) {
+                        for (int i = 0; i < headers.size(); i++) {
+                            System.out.println(connection.getRemoteAddress() + " (" + streamId + ") << " + headers.get(i));
+                        }
+                    }
+
+                    @Override
+                    public void onHeaderOutput(final HttpConnection connection, final int streamId, final List<? extends Header> headers) {
+                        for (int i = 0; i < headers.size(); i++) {
+                            System.out.println(connection.getRemoteAddress() + " (" + streamId + ") >> " + headers.get(i));
+                        }
+                    }
+
+                    @Override
+                    public void onFrameInput(final HttpConnection connection, final int streamId, final RawFrame frame) {
+                    }
+
+                    @Override
+                    public void onFrameOutput(final HttpConnection connection, final int streamId, final RawFrame frame) {
+                    }
+
+                    @Override
+                    public void onInputFlowControl(final HttpConnection connection, final int streamId, final int delta, final int actualSize) {
+                    }
+
+                    @Override
+                    public void onOutputFlowControl(final HttpConnection connection, final int streamId, final int delta, final int actualSize) {
+                    }
+
+                })
+                .create();
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            System.out.println("HTTP requester shutting down");
+            requester.close(CloseMode.GRACEFUL);
+        }));
+        requester.start();
+
+        final HttpHost target = new HttpHost("nghttp2.org");
+        final Future<AsyncClientEndpoint> future = requester.connect(target, Timeout.ofDays(5));
+        final AsyncClientEndpoint clientEndpoint = future.get();
+
+        final String[] requestUris = new String[] {"/httpbin/ip", "/httpbin/user-agent", "/httpbin/headers"};
+
+        for (final String requestUri: requestUris) {
+            final ClassicHttpRequest request = ClassicRequestBuilder.get()
+                    .setHttpHost(target)
+                    .setPath(requestUri)
+                    .build();
+
+            final ClassicToAsyncRequestProducer requestProducer = new ClassicToAsyncRequestProducer(request, Timeout.ofMinutes(5));
+            final ClassicToAsyncResponseConsumer responseConsumer = new ClassicToAsyncResponseConsumer(Timeout.ofMinutes(5));
+
+            clientEndpoint.execute(requestProducer, responseConsumer, null);
+
+            requestProducer.blockWaiting().execute();
+            try (ClassicHttpResponse response = responseConsumer.blockWaiting()) {
+                System.out.println(requestUri + " -> " + response.getCode());
+                final HttpEntity entity = response.getEntity();
+                if (entity != null) {
+                    final ContentType contentType = ContentType.parse(entity.getContentType());
+                    final Charset charset = ContentType.getCharset(contentType, StandardCharsets.UTF_8);
+                    try (final BufferedReader reader = new BufferedReader(new InputStreamReader(entity.getContent(), charset))) {
+                        String line;
+                        while ((line = reader.readLine()) != null) {
+                            System.out.println(line);
+                        }
+                    }
+                }
+            }
+        }
+
+        System.out.println("Shutting down I/O reactor");
+        requester.initiateShutdown();
+    }
+
+}