ProxyConnectConnectionFactoryFilter leaks connection in case of errors

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

@@ -1,5 +1,5 @@
 /*
- * Copyright © 2018-2019 Apple Inc. and the ServiceTalk project authors
+ * Copyright © 2018-2020 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.
@@ -244,7 +244,7 @@ private static <U, R> StreamingHttpClient buildStreaming(final HttpClientBuildCo
             if (roConfig.hasProxy() && sslContext != null) {
                 assert roConfig.connectAddress() != null;
                 connectionFactoryFilter = new ProxyConnectConnectionFactoryFilter<R, FilterableStreamingHttpConnection>(
-                        roConfig.connectAddress(), reqRespFactory).append(connectionFactoryFilter);
+                        roConfig.connectAddress()).append(connectionFactoryFilter);
             }
 
             final HttpExecutionStrategy executionStrategy = ctx.executionContext.executionStrategy();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright © 2019 Apple Inc. and the ServiceTalk project authors
+ * Copyright © 2019-2020 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.
@@ -19,12 +19,11 @@
 import io.servicetalk.client.api.ConnectionFactoryFilter;
 import io.servicetalk.client.api.DelegatingConnectionFactory;
 import io.servicetalk.concurrent.SingleSource;
-import io.servicetalk.concurrent.api.ListenableAsyncCloseable;
 import io.servicetalk.concurrent.api.Single;
 import io.servicetalk.http.api.FilterableStreamingHttpConnection;
 import io.servicetalk.http.api.HttpExecutionStrategy;
 import io.servicetalk.http.api.HttpExecutionStrategyInfluencer;
-import io.servicetalk.http.api.StreamingHttpRequestResponseFactory;
+import io.servicetalk.http.api.StreamingHttpResponse;
 import io.servicetalk.transport.netty.internal.DeferSslHandler;
 import io.servicetalk.transport.netty.internal.NettyConnectionContext;
 
@@ -47,17 +46,12 @@
  * @param <ResolvedAddress> The type of a resolved address that can be used for connecting.
  * @param <C> The type of connections created by this factory.
  */
-final class ProxyConnectConnectionFactoryFilter<ResolvedAddress, C
-        extends ListenableAsyncCloseable & FilterableStreamingHttpConnection>
-        implements ConnectionFactoryFilter<ResolvedAddress, C>,
-                   HttpExecutionStrategyInfluencer {
+final class ProxyConnectConnectionFactoryFilter<ResolvedAddress, C extends FilterableStreamingHttpConnection>
+        implements ConnectionFactoryFilter<ResolvedAddress, C>, HttpExecutionStrategyInfluencer {
 
-    private final StreamingHttpRequestResponseFactory reqRespFactory;
     private final String connectAddress;
 
-    ProxyConnectConnectionFactoryFilter(final CharSequence connectAddress,
-                                        final StreamingHttpRequestResponseFactory reqRespFactory) {
-        this.reqRespFactory = reqRespFactory;
+    ProxyConnectConnectionFactoryFilter(final CharSequence connectAddress) {
         this.connectAddress = connectAddress.toString();
     }
 
@@ -74,51 +68,59 @@ private ProxyFilter(final ConnectionFactory<ResolvedAddress, C> delegate) {
 
         @Override
         public Single<C> newConnection(final ResolvedAddress resolvedAddress) {
-            return delegate().newConnection(resolvedAddress).flatMap(c ->
-                // We currently only have access to a StreamingHttpRequester, which means we are forced to provide an
-                // HttpExecutionStrategy. Because we can't be sure if there is any blocking code in the connection
-                // filters we use the default strategy which should offload everything to be safe.
-                c.request(defaultStrategy(),
-                            reqRespFactory.connect(connectAddress).addHeader(CONTENT_LENGTH, ZERO))
-                 .flatMap(response -> {
-                if (SUCCESSFUL_2XX.contains(response.status())) {
-                    final Channel channel = ((NettyConnectionContext) c.connectionContext()).nettyChannel();
-                    final SingleSource.Processor<C, C> processor = newSingleProcessor();
+            return delegate().newConnection(resolvedAddress).flatMap(c -> {
+                try {
+                    // We currently only have access to a StreamingHttpRequester, which means we are forced to provide
+                    // an HttpExecutionStrategy. Because we can't be sure if there is any blocking code in the
+                    // connection filters we use the default strategy which should offload everything to be safe.
+                    return c.request(defaultStrategy(), c.connect(connectAddress).addHeader(CONTENT_LENGTH, ZERO))
+                            .flatMap(response -> handleConnectResponse(c, response))
+                            // Close recently created connection in case of any error while it connects to the proxy:
+                            .recoverWith(t -> c.closeAsync().concat(failed(t)));
+                } catch (Throwable t) {
+                    return c.closeAsync().concat(failed(t));
+                }
+            });
+        }
+    }
 
-                    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(c);
-                                } else {
-                                    processor.onError(event.cause());
-                                }
-                            }
-                            ctx.fireUserEventTriggered(evt);
-                        }
-                    });
+    private Single<C> handleConnectResponse(final C connection, final StreamingHttpResponse response) {
+        try {
+            if (response.status().statusClass() != SUCCESSFUL_2XX) {
+                return failed(new ProxyResponseException("Non-successful response from proxy CONNECT " +
+                        connectAddress, response.status()));
+            }
 
-                    DeferSslHandler deferSslHandler = channel.pipeline().get(DeferSslHandler.class);
-                    if (deferSslHandler == null) {
-                        return response.payloadBodyAndTrailers().ignoreElements().concat(failed(
-                                new IllegalStateException("Failed to find a handler of type " +
-                                        DeferSslHandler.class + " in channel pipeline.")));
+            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);
+                }
+            });
 
-                    deferSslHandler.ready();
+            final DeferSslHandler deferSslHandler = channel.pipeline().get(DeferSslHandler.class);
+            if (deferSslHandler == null) {
+                return failed(new IllegalStateException("Failed to find a handler of type " +
+                        DeferSslHandler.class + " in channel pipeline."));
+            }
+            deferSslHandler.ready();
 
-                    // There is no need to apply offloading explicitly (despite completing `processor` on the
-                    // EventLoop) because `payloadBody()` will be offloaded according to the strategy for the
-                    // request.
-                    return response.payloadBodyAndTrailers().ignoreElements().concat(fromSource(processor));
-                } else {
-                    return response.payloadBodyAndTrailers().ignoreElements().concat(
-                            failed(new ProxyResponseException("Bad response from proxy CONNECT " + connectAddress,
-                                    response.status())));
-                }
-            }));
+            // There is no need to apply offloading explicitly (despite completing `processor` on the
+            // EventLoop) because `payloadBody()` will be offloaded according to the strategy for the
+            // request.
+            return response.payloadBodyAndTrailers().ignoreElements().concat(fromSource(processor));
+        } catch (Throwable t) {
+            return failed(t);
         }
     }