Move HTTP proxy CONNECT logic before ConnectionFactoryFilters

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

@@ -49,11 +49,16 @@
 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());
     }

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.
@@ -278,14 +278,17 @@ 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);
             } else {
-                H1ProtocolConfig h1Config = roConfig.h1Config();
-                assert h1Config != null;
                 connectionFactory = new PipelinedLBHttpConnectionFactory<>(roConfig, executionContext,
                         connectionFilterFactory, reqRespFactory,
                         connectionFactoryStrategy, connectionFactoryFilter,

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

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

@@ -1,5 +1,5 @@
 /*
- * Copyright © 2019-2020, 2022-2023 Apple Inc. and the ServiceTalk project authors
+ * Copyright © 2019-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.
@@ -18,40 +18,31 @@
 import io.servicetalk.client.api.ConnectionFactory;
 import io.servicetalk.client.api.ConnectionFactoryFilter;
 import io.servicetalk.client.api.DelegatingConnectionFactory;
-import io.servicetalk.concurrent.SingleSource;
 import io.servicetalk.concurrent.api.Single;
 import io.servicetalk.concurrent.internal.DefaultContextMap;
 import io.servicetalk.context.api.ContextMap;
 import io.servicetalk.http.api.FilterableStreamingHttpConnection;
+import io.servicetalk.http.api.HttpContextKeys;
 import io.servicetalk.http.api.HttpExecutionStrategies;
 import io.servicetalk.http.api.HttpExecutionStrategy;
-import io.servicetalk.http.api.StreamingHttpResponse;
-import io.servicetalk.transport.api.ConnectExecutionStrategy;
 import io.servicetalk.transport.api.ExecutionStrategy;
-import io.servicetalk.transport.api.IoThreadFactory;
 import io.servicetalk.transport.api.TransportObserver;
-import io.servicetalk.transport.netty.internal.DeferSslHandler;
-import io.servicetalk.transport.netty.internal.NettyConnectionContext;
-import io.servicetalk.transport.netty.internal.StacklessClosedChannelException;
 
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInboundHandlerAdapter;
-import io.netty.handler.ssl.SslHandshakeCompletionEvent;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nullable;
 
-import static io.servicetalk.concurrent.api.Processors.newSingleProcessor;
-import static io.servicetalk.concurrent.api.Single.failed;
-import static io.servicetalk.concurrent.api.SourceAdapters.fromSource;
 import static io.servicetalk.http.api.HttpContextKeys.HTTP_TARGET_ADDRESS_BEHIND_PROXY;
-import static io.servicetalk.http.api.HttpHeaderNames.HOST;
-import static io.servicetalk.http.api.HttpResponseStatus.StatusClass.SUCCESSFUL_2XX;
 
 /**
- * A connection factory filter that sends a `CONNECT` request for https proxying.
+ * A {@link ConnectionFactoryFilter} that is prepended before any user-defined filters for the purpose of setting a
+ * {@link HttpContextKeys#HTTP_TARGET_ADDRESS_BEHIND_PROXY} key.
+ * <p>
+ * The actual logic to do a proxy connect was moved to {@link ProxyConnectLBHttpConnectionFactory}.
+ * <p>
+ * This filter can be removed when {@link HttpContextKeys#HTTP_TARGET_ADDRESS_BEHIND_PROXY} key is deprecated and
+ * removed.
  *
  * @param <ResolvedAddress> The type of resolved addresses that can be used for connecting.
  * @param <C> The type of connections created by this factory.
@@ -62,12 +53,9 @@ final class ProxyConnectConnectionFactoryFilter<ResolvedAddress, C extends Filte
     private static final Logger LOGGER = LoggerFactory.getLogger(ProxyConnectConnectionFactoryFilter.class);
 
     private final String connectAddress;
-    private final boolean isConnectOffloaded;
 
     ProxyConnectConnectionFactoryFilter(final CharSequence connectAddress, final ExecutionStrategy connectStrategy) {
         this.connectAddress = connectAddress.toString();
-        this.isConnectOffloaded = connectStrategy instanceof ConnectExecutionStrategy &&
-                ((ConnectExecutionStrategy) connectStrategy).isConnectOffloaded();
     }
 
     @Override
@@ -89,77 +77,11 @@ public Single<C> newConnection(final ResolvedAddress resolvedAddress,
                 final ContextMap contextMap = context != null ? context : new DefaultContextMap();
                 logUnexpectedAddress(contextMap.put(HTTP_TARGET_ADDRESS_BEHIND_PROXY, connectAddress),
                         connectAddress, LOGGER);
-                return delegate().newConnection(resolvedAddress, contextMap, observer).flatMap(c -> {
-                    try {
-                        // Send CONNECT request: https://datatracker.ietf.org/doc/html/rfc9110#section-9.3.6
-                        // Host header value must be equal to CONNECT request target, see
-                        // https://github.com/haproxy/haproxy/issues/1159
-                        // https://datatracker.ietf.org/doc/html/rfc7230#section-5.4:
-                        //   If the target URI includes an authority component, then a client MUST send a field-value
-                        //   for Host that is identical to that authority component
-                        return c.request(c.connect(connectAddress).setHeader(HOST, connectAddress))
-                                // Successful response to CONNECT never has a message body, and we are not interested in
-                                // payload body for any non-200 status code. Drain it asap to free connection and RS
-                                // resources before starting TLS handshake.
-                                .flatMap(response -> response.messageBody().ignoreElements()
-                                        .concat(Single.defer(() -> handleConnectResponse(c, response)
-                                                .shareContextOnSubscribe()))
-                                        .shareContextOnSubscribe())
-                                // Close recently created connection in case of any error while it connects to the
-                                // proxy:
-                                .onErrorResume(t -> c.closeAsync().concat(failed(t)));
-                        // We do not apply shareContextOnSubscribe() here to isolate a context for `CONNECT` request.
-                    } catch (Throwable t) {
-                        return c.closeAsync().concat(failed(t));
-                    }
-                }).shareContextOnSubscribe();
+                // The rest of the logic was moved to ProxyConnectLBHttpConnectionFactory
+                return delegate().newConnection(resolvedAddress, contextMap, observer)
+                        .shareContextOnSubscribe();
             });
         }
-
-        private Single<C> handleConnectResponse(final C connection, final StreamingHttpResponse response) {
-            if (response.status().statusClass() != SUCCESSFUL_2XX) {
-                return failed(new ProxyResponseException(connection + " Non-successful response from proxy CONNECT " +
-                        connectAddress, response.status()));
-            }
-
-            final Channel channel = ((NettyConnectionContext) connection.connectionContext()).nettyChannel();
-            final SingleSource.Processor<C, C> processor = newSingleProcessor();
-            channel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
-                @Override
-                public void userEventTriggered(final ChannelHandlerContext ctx, final Object evt) {
-                    if (evt instanceof SslHandshakeCompletionEvent) {
-                        SslHandshakeCompletionEvent event = (SslHandshakeCompletionEvent) evt;
-                        if (event.isSuccess()) {
-                            processor.onSuccess(connection);
-                        } else {
-                            processor.onError(event.cause());
-                        }
-                    }
-                    ctx.fireUserEventTriggered(evt);
-                }
-            });
-
-            final DeferSslHandler deferSslHandler = channel.pipeline().get(DeferSslHandler.class);
-            if (deferSslHandler == null) {
-                if (!channel.isActive()) {
-                    LOGGER.info("{} is unexpectedly closed after receiving response: {}. " +
-                                    "Investigate logs on a proxy side to identify the cause.",
-                            connection, response.toString((name, value) -> value));
-                    return failed(StacklessClosedChannelException.newInstance(
-                            ProxyConnectConnectionFactoryFilter.class, "handleConnectResponse: " +
-                                    connection + " is unexpectedly closed. Check logs for more info."));
-                }
-                return failed(new IllegalStateException(connection + " Failed to find a handler of type " +
-                        DeferSslHandler.class + " in channel pipeline."));
-            }
-            deferSslHandler.ready();
-
-            // processor completes on EventLoop thread, apply offloading if required:
-            return isConnectOffloaded ?
-                    fromSource(processor).publishOn(connection.executionContext().executor(),
-                            IoThreadFactory.IoThread::currentThreadIsIoThread) :
-                    fromSource(processor);
-        }
     }
 
     static void logUnexpectedAddress(@Nullable final Object current, final Object expected, final Logger logger) {