Add initializer API for HTTP proxy CONNECT request

servicetalk-http-api/src/main/java/io/servicetalk/http/api/DelegatingSingleAddressHttpClientBuilder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright © 2022 Apple Inc. and the ServiceTalk project authors
+ * Copyright © 2022-2023 Apple Inc. and the ServiceTalk project authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -28,6 +28,7 @@
 
 import java.net.SocketOption;
 import java.util.function.BooleanSupplier;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
 
@@ -73,6 +74,13 @@ public SingleAddressHttpClientBuilder<U, R> proxyAddress(final U proxyAddress) {
         return this;
     }
 
+    @Override
+    public SingleAddressHttpClientBuilder<U, R> proxyAddress(
+            final U proxyAddress, final Consumer<StreamingHttpRequest> connectRequestInitializer) {
+        delegate = delegate.proxyAddress(proxyAddress, connectRequestInitializer);
+        return this;
+    }
+
     @Override
     public <T> SingleAddressHttpClientBuilder<U, R> socketOption(final SocketOption<T> option, final T value) {
         delegate = delegate.socketOption(option, value);

servicetalk-http-api/src/main/java/io/servicetalk/http/api/SingleAddressHttpClientBuilder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright © 2018-2019 Apple Inc. and the ServiceTalk project authors
+ * Copyright © 2018-2023 Apple Inc. and the ServiceTalk project authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -34,6 +34,7 @@
 import java.net.SocketOption;
 import java.net.StandardSocketOptions;
 import java.util.function.BooleanSupplier;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
 
@@ -49,15 +50,40 @@
 public interface SingleAddressHttpClientBuilder<U, R> extends HttpClientBuilder<U, R, ServiceDiscovererEvent<R>> {
     /**
      * Configure proxy to serve as an intermediary for requests.
+     * <p>
+     * If the client talks to a proxy over http (not https, {@link #sslConfig(ClientSslConfig) ClientSslConfig} is NOT
+     * configured), it will rewrite the request-target to
+     * <a href="https://tools.ietf.org/html/rfc7230#section-5.3.2">absolute-form</a>, as specified by the RFC.
+     *
      * @param proxyAddress Unresolved address of the proxy. When used with a builder created for a resolved address,
      * {@code proxyAddress} should also be already resolved – otherwise runtime exceptions may occur.
      * @return {@code this}.
      */
-    default SingleAddressHttpClientBuilder<U, R> proxyAddress(U proxyAddress) {
+    default SingleAddressHttpClientBuilder<U, R> proxyAddress(U proxyAddress) { // FIXME: 0.43 - remove default impl
         throw new UnsupportedOperationException("Setting proxy address is not yet supported by "
                 + getClass().getName());
     }
 
+    /**
+     * Configure proxy to serve as an intermediary for requests.
+     * <p>
+     * If the client talks to a proxy over http (not https, {@link #sslConfig(ClientSslConfig) ClientSslConfig} is NOT
+     * configured), it will rewrite the request-target to
+     * <a href="https://tools.ietf.org/html/rfc7230#section-5.3.2">absolute-form</a>, as specified by the RFC.
+     *
+     * @param proxyAddress Unresolved address of the proxy. When used with a builder created for a resolved address,
+     * {@code proxyAddress} should also be already resolved – otherwise runtime exceptions may occur.
+     * @param connectRequestInitializer {@link Consumer} of {@link StreamingHttpRequest} that can be used to add
+     * additional info to <a href="https://datatracker.ietf.org/doc/html/rfc9110#section-9.3.6">HTTP/1.1 CONNECT</a>
+     * request. It can be used to add headers, like {@link HttpHeaderNames#PROXY_AUTHORIZATION}, debugging info, etc.
+     * @return {@code this}.
+     */
+    default SingleAddressHttpClientBuilder<U, R> proxyAddress(// FIXME: 0.43 - remove default impl
+            U proxyAddress, Consumer<StreamingHttpRequest> connectRequestInitializer) {
+        throw new UnsupportedOperationException(
+                "Setting proxy address with request initializer is not yet supported by " + getClass().getName());
+    }
+
     /**
      * Adds a {@link SocketOption} for all connections created by this builder.
      *

servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/AbstractStreamingHttpConnection.java

@@ -1,5 +1,5 @@
 /*
- * Copyright © 2018-2019, 2021-2022 Apple Inc. and the ServiceTalk project authors
+ * Copyright © 2018-2023 Apple Inc. and the ServiceTalk project authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,7 +20,6 @@
 import io.servicetalk.concurrent.api.Publisher;
 import io.servicetalk.concurrent.api.Single;
 import io.servicetalk.concurrent.api.TerminalSignalConsumer;
-import io.servicetalk.http.api.FilterableStreamingHttpConnection;
 import io.servicetalk.http.api.HttpConnectionContext;
 import io.servicetalk.http.api.HttpEventKey;
 import io.servicetalk.http.api.HttpExecutionContext;
@@ -39,6 +38,7 @@
 import io.servicetalk.transport.netty.internal.FlushStrategy;
 import io.servicetalk.transport.netty.internal.NettyConnectionContext;
 
+import io.netty.channel.Channel;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -66,7 +66,7 @@
 import static java.util.Objects.requireNonNull;
 
 abstract class AbstractStreamingHttpConnection<CC extends NettyConnectionContext>
-        implements FilterableStreamingHttpConnection {
+        implements NettyFilterableStreamingHttpConnection {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(AbstractStreamingHttpConnection.class);
     static final IgnoreConsumedEvent<Integer> ZERO_MAX_CONCURRENCY_EVENT = new IgnoreConsumedEvent<>(0);
@@ -168,7 +168,7 @@ public void cancel() {
     }
 
     @Override
-    public Single<StreamingHttpResponse> request(final StreamingHttpRequest request) {
+    public final Single<StreamingHttpResponse> request(final StreamingHttpRequest request) {
         return defer(() -> {
             Publisher<Object> flatRequest;
             // See https://tools.ietf.org/html/rfc7230#section-3.3.3
@@ -260,6 +260,11 @@ public final StreamingHttpResponseFactory httpResponseFactory() {
         return reqRespFactory;
     }
 
+    @Override
+    public Channel nettyChannel() {
+        return connection.nettyChannel();
+    }
+
     @Override
     public final Completable onClose() {
         return connectionContext.onClose();

servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/DefaultSingleAddressHttpClientBuilder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright © 2018-2022 Apple Inc. and the ServiceTalk project authors
+ * Copyright © 2018-2023 Apple Inc. and the ServiceTalk project authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -65,6 +65,7 @@
 import java.time.Duration;
 import java.util.Collection;
 import java.util.function.BooleanSupplier;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
 import javax.annotation.Nullable;
@@ -109,6 +110,7 @@ final class DefaultSingleAddressHttpClientBuilder<U, R> implements SingleAddress
     private final U address;
     @Nullable
     private U proxyAddress;
+    private Consumer<StreamingHttpRequest> proxyConnectRequestInitializer = __ -> { };
     private final HttpClientConfig config;
     final HttpExecutionContextBuilder executionContextBuilder;
     private final ClientStrategyInfluencerChainBuilder strategyComputation;
@@ -146,6 +148,7 @@ private DefaultSingleAddressHttpClientBuilder(@Nullable final U address,
                                                   final DefaultSingleAddressHttpClientBuilder<U, R> from) {
         this.address = address;
         proxyAddress = from.proxyAddress;
+        proxyConnectRequestInitializer = from.proxyConnectRequestInitializer;
         config = new HttpClientConfig(from.config);
         executionContextBuilder = new HttpExecutionContextBuilder(from.executionContextBuilder);
         strategyComputation = from.strategyComputation.copy();
@@ -278,14 +281,18 @@ public HttpExecutionStrategy executionStrategy() {
                 H2ProtocolConfig h2Config = roConfig.h2Config();
                 connectionFactory = new AlpnLBHttpConnectionFactory<>(roConfig, executionContext,
                         connectionFilterFactory, new AlpnReqRespFactoryFunc(
-                                executionContext.bufferAllocator(),
-                                h1Config == null ? null : h1Config.headersFactory(),
-                                h2Config == null ? null : h2Config.headersFactory()),
+                        executionContext.bufferAllocator(),
+                        h1Config == null ? null : h1Config.headersFactory(),
+                        h2Config == null ? null : h2Config.headersFactory()),
                         connectionFactoryStrategy, connectionFactoryFilter,
                         ctx.builder.loadBalancerFactory::toLoadBalancedConnection);
+            } else if (roConfig.hasProxy() && sslContext != null) {
+                connectionFactory = new ProxyConnectLBHttpConnectionFactory<>(roConfig, executionContext,
+                        connectionFilterFactory, reqRespFactory,
+                        connectionFactoryStrategy, connectionFactoryFilter,
+                        ctx.builder.loadBalancerFactory::toLoadBalancedConnection,
+                        ctx.builder.proxyConnectRequestInitializer);
             } else {
-                H1ProtocolConfig h1Config = roConfig.h1Config();
-                assert h1Config != null;
                 connectionFactory = new PipelinedLBHttpConnectionFactory<>(roConfig, executionContext,
                         connectionFilterFactory, reqRespFactory,
                         connectionFactoryStrategy, connectionFactoryFilter,
@@ -446,6 +453,14 @@ public DefaultSingleAddressHttpClientBuilder<U, R> proxyAddress(final U proxyAdd
         return this;
     }
 
+    @Override
+    public SingleAddressHttpClientBuilder<U, R> proxyAddress(
+            final U proxyAddress, final Consumer<StreamingHttpRequest> connectRequestInitializer) {
+        this.proxyAddress(proxyAddress);
+        this.proxyConnectRequestInitializer = requireNonNull(connectRequestInitializer);
+        return this;
+    }
+
     @Override
     public DefaultSingleAddressHttpClientBuilder<U, R> ioExecutor(final IoExecutor ioExecutor) {
         executionContextBuilder.ioExecutor(ioExecutor);

servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/NettyFilterableStreamingHttpConnection.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright © 2023 Apple Inc. and the ServiceTalk project authors
+ *
+ * Licensed 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 io.servicetalk.http.netty;
+
+import io.servicetalk.http.api.FilterableStreamingHttpConnection;
+
+import io.netty.channel.Channel;
+
+/**
+ * {@link FilterableStreamingHttpConnection} that also gives access to Netty {@link Channel}.
+ */
+interface NettyFilterableStreamingHttpConnection extends FilterableStreamingHttpConnection {
+
+    /**
+     * Return the Netty {@link Channel} backing this connection.
+     *
+     * @return the Netty {@link Channel} backing this connection.
+     */
+    Channel nettyChannel();
+}

servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/PipelinedLBHttpConnectionFactory.java

@@ -28,6 +28,7 @@
 
 import static io.servicetalk.http.api.HttpProtocolVersion.HTTP_1_1;
 import static io.servicetalk.http.netty.StreamingConnectionFactory.buildStreaming;
+import static java.util.Objects.requireNonNull;
 
 final class PipelinedLBHttpConnectionFactory<ResolvedAddress> extends AbstractLBHttpConnectionFactory<ResolvedAddress> {
     PipelinedLBHttpConnectionFactory(
@@ -39,6 +40,7 @@ final class PipelinedLBHttpConnectionFactory<ResolvedAddress> extends AbstractLB
             final ProtocolBinding protocolBinding) {
         super(config, executionContext, version -> reqRespFactory, connectStrategy, connectionFactoryFilter,
                 connectionFilterFunction, protocolBinding);
+        requireNonNull(config.h1Config(), "H1ProtocolConfig is required");
     }
 
     @Override