Just keep minimal changes

Author: arturobernalg

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

@@ -0,0 +1,73 @@
+/*
+ * ====================================================================
+ * 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.impl.nio.bootstrap;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.hc.core5.concurrent.Cancellable;
+import org.apache.hc.core5.concurrent.CancellableDependency;
+
+final class CancellableExecution implements CancellableDependency {
+
+    private final AtomicBoolean cancelled;
+    private final AtomicReference<Cancellable> dependencyRef;
+
+    CancellableExecution() {
+        this.cancelled = new AtomicBoolean();
+        this.dependencyRef = new AtomicReference<>();
+    }
+
+    @Override
+    public void setDependency(final Cancellable cancellable) {
+        dependencyRef.set(cancellable);
+        if (cancelled.get()) {
+            final Cancellable dependency = dependencyRef.getAndSet(null);
+            if (dependency != null) {
+                dependency.cancel();
+            }
+        }
+    }
+
+    @Override
+    public boolean isCancelled() {
+        return cancelled.get();
+    }
+
+    @Override
+    public boolean cancel() {
+        if (cancelled.compareAndSet(false, true)) {
+            final Cancellable dependency = dependencyRef.getAndSet(null);
+            if (dependency != null) {
+                dependency.cancel();
+            }
+            return true;
+        }
+        return false;
+    }
+
+}

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

@@ -29,6 +29,8 @@
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.Future;
 import java.util.concurrent.RejectedExecutionException;
@@ -43,15 +45,21 @@
 import org.apache.hc.core5.function.Decorator;
 import org.apache.hc.core5.function.Resolver;
 import org.apache.hc.core5.http.ConnectionClosedException;
+import org.apache.hc.core5.http.EntityDetails;
+import org.apache.hc.core5.http.Header;
 import org.apache.hc.core5.http.HttpException;
 import org.apache.hc.core5.http.HttpHost;
+import org.apache.hc.core5.http.HttpResponse;
 import org.apache.hc.core5.http.impl.DefaultAddressResolver;
 import org.apache.hc.core5.http.impl.bootstrap.AsyncRequester;
 import org.apache.hc.core5.http.nio.AsyncClientExchangeHandler;
 import org.apache.hc.core5.http.nio.AsyncPushConsumer;
 import org.apache.hc.core5.http.nio.AsyncRequestProducer;
 import org.apache.hc.core5.http.nio.AsyncResponseConsumer;
+import org.apache.hc.core5.http.nio.CapacityChannel;
+import org.apache.hc.core5.http.nio.DataStreamChannel;
 import org.apache.hc.core5.http.nio.HandlerFactory;
+import org.apache.hc.core5.http.nio.RequestChannel;
 import org.apache.hc.core5.http.nio.command.RequestExecutionCommand;
 import org.apache.hc.core5.http.nio.command.ShutdownCommand;
 import org.apache.hc.core5.http.nio.ssl.TlsStrategy;
@@ -183,14 +191,69 @@ private void execute(
                 if (request.getAuthority() == null) {
                     request.setAuthority(new URIAuthority(host));
                 }
-                if (request.getScheme() == null) {
-                    request.setScheme(host.getSchemeName());
-                }
                 connPool.getSession(host, timeout, new FutureCallback<IOSession>() {
 
                     @Override
                     public void completed(final IOSession ioSession) {
+                        final AsyncClientExchangeHandler handlerProxy = new AsyncClientExchangeHandler() {
+
+                            @Override
+                            public void releaseResources() {
+                                exchangeHandler.releaseResources();
+                            }
+
+                            @Override
+                            public void produceRequest(final RequestChannel channel, final HttpContext httpContext) throws HttpException, IOException {
+                                channel.sendRequest(request, entityDetails, httpContext);
+                            }
+
+                            @Override
+                            public int available() {
+                                return exchangeHandler.available();
+                            }
+
+                            @Override
+                            public void produce(final DataStreamChannel channel) throws IOException {
+                                exchangeHandler.produce(channel);
+                            }
+
+                            @Override
+                            public void consumeInformation(final HttpResponse response, final HttpContext httpContext) throws HttpException, IOException {
+                                exchangeHandler.consumeInformation(response, httpContext);
+                            }
+
+                            @Override
+                            public void consumeResponse(
+                                    final HttpResponse response, final EntityDetails entityDetails, final HttpContext httpContext) throws HttpException, IOException {
+                                exchangeHandler.consumeResponse(response, entityDetails, httpContext);
+                            }
+
+                            @Override
+                            public void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
+                                exchangeHandler.updateCapacity(capacityChannel);
+                            }
+
+                            @Override
+                            public void consume(final ByteBuffer src) throws IOException {
+                                exchangeHandler.consume(src);
+                            }
+
+                            @Override
+                            public void streamEnd(final List<? extends Header> trailers) throws HttpException, IOException {
+                                exchangeHandler.streamEnd(trailers);
+                            }
 
+                            @Override
+                            public void cancel() {
+                                exchangeHandler.cancel();
+                            }
+
+                            @Override
+                            public void failed(final Exception cause) {
+                                exchangeHandler.failed(cause);
+                            }
+
+                        };
                         final int max = maxRequestsPerConnection;
                         if (max > 0) {
                             final int current = ioSession.getPendingCommandCount();
@@ -201,24 +264,20 @@ public void completed(final IOSession ioSession) {
                                 return;
                             }
                         }
-
                         final Timeout socketTimeout = ioSession.getSocketTimeout();
-                        final RequestExecutionCommand command = new RequestExecutionCommand(
-                                exchangeHandler,
-                                pushHandlerFactory,
-                                context,
-                                streamControl -> {
-                                    cancellableDependency.setDependency(streamControl);
-                                    if (socketTimeout != null) {
-                                        streamControl.setTimeout(socketTimeout);
-                                    }
-                                });
-
-                        ioSession.enqueue(command, Command.Priority.NORMAL);
-
+                        ioSession.enqueue(new RequestExecutionCommand(
+                                        handlerProxy,
+                                        pushHandlerFactory,
+                                        context,
+                                        streamControl -> {
+                                            cancellableDependency.setDependency(streamControl);
+                                            if (socketTimeout != null) {
+                                                streamControl.setTimeout(socketTimeout);
+                                            }
+                                        }),
+                                Command.Priority.NORMAL);
                         if (!ioSession.isOpen()) {
                             exchangeHandler.failed(new ConnectionClosedException());
-                            exchangeHandler.releaseResources();
                         }
                     }
 
@@ -309,34 +368,4 @@ public final <T> Future<T> execute(
     public H2ConnPool getConnPool() {
         return connPool;
     }
-
-    /**
-     * Cancellable that can be wired to the stream control once it becomes available.
-     */
-    private static final class CancellableExecution implements Cancellable, CancellableDependency {
-
-        private volatile Cancellable dependency;
-
-        @Override
-        public void setDependency(final Cancellable dependency) {
-            this.dependency = dependency;
-        }
-
-        @Override
-        public boolean isCancelled() {
-            return false;
-        }
-
-        @Override
-        public boolean cancel() {
-            final Cancellable local = this.dependency;
-            if (local != null) {
-                local.cancel();
-                return true;
-            }
-            return false;
-        }
-
-    }
-
 }