http-proxy: attach headers to connection exception (#8824)

Motivation:
When a proxy fails to connect, it includes useful error detail in
the headers.

Modification:
- Add an HTTP Specific ProxyConnectException
- Attach headers (if any) in the event of a non-200 response

Result:
Able to surface more useful error info to applications

handler-proxy/src/main/java/io/netty/handler/proxy/HttpProxyHandler.java

@@ -47,9 +47,10 @@ public final class HttpProxyHandler extends ProxyHandler {
     private final String username;
     private final String password;
     private final CharSequence authorization;
+    private final HttpHeaders outboundHeaders;
     private final boolean ignoreDefaultPortsInConnectHostHeader;
     private HttpResponseStatus status;
-    private HttpHeaders headers;
+    private HttpHeaders inboundHeaders;
 
     public HttpProxyHandler(SocketAddress proxyAddress) {
         this(proxyAddress, null);
@@ -66,7 +67,7 @@ public final class HttpProxyHandler extends ProxyHandler {
         username = null;
         password = null;
         authorization = null;
-        this.headers = headers;
+        this.outboundHeaders = headers;
         this.ignoreDefaultPortsInConnectHostHeader = ignoreDefaultPortsInConnectHostHeader;
     }
 
@@ -102,7 +103,7 @@ public final class HttpProxyHandler extends ProxyHandler {
         authz.release();
         authzBase64.release();
 
-        this.headers = headers;
+        this.outboundHeaders = headers;
         this.ignoreDefaultPortsInConnectHostHeader = ignoreDefaultPortsInConnectHostHeader;
     }
 
@@ -163,32 +164,59 @@ public final class HttpProxyHandler extends ProxyHandler {
             req.headers().set(HttpHeaderNames.PROXY_AUTHORIZATION, authorization);
         }
 
-        if (headers != null) {
-            req.headers().add(headers);
+        if (outboundHeaders != null) {
+            req.headers().add(outboundHeaders);
         }
 
         return req;
     }
 
     @Override
-    protected boolean handleResponse(ChannelHandlerContext ctx, Object response) throws Exception {
+    protected boolean handleResponse(ChannelHandlerContext ctx, Object response) throws HttpProxyConnectException {
         if (response instanceof HttpResponse) {
             if (status != null) {
-                throw new ProxyConnectException(exceptionMessage("too many responses"));
+                throw new HttpProxyConnectException(exceptionMessage("too many responses"), /*headers=*/ null);
             }
-            status = ((HttpResponse) response).status();
+            HttpResponse res = (HttpResponse) response;
+            status = res.status();
+            inboundHeaders = res.headers();
         }
 
         boolean finished = response instanceof LastHttpContent;
         if (finished) {
             if (status == null) {
-                throw new ProxyConnectException(exceptionMessage("missing response"));
+                throw new HttpProxyConnectException(exceptionMessage("missing response"), inboundHeaders);
             }
             if (status.code() != 200) {
-                throw new ProxyConnectException(exceptionMessage("status: " + status));
+                throw new HttpProxyConnectException(exceptionMessage("status: " + status), inboundHeaders);
             }
         }
 
         return finished;
     }
+
+    /**
+     * Specific case of a connection failure, which may include headers from the proxy.
+     */
+    public static final class HttpProxyConnectException extends ProxyConnectException {
+        private static final long serialVersionUID = -8824334609292146066L;
+
+        private final HttpHeaders headers;
+
+        /**
+         * @param message The failure message.
+         * @param headers Header associated with the connection failure.  May be {@code null}.
+         */
+        public HttpProxyConnectException(String message, HttpHeaders headers) {
+            super(message);
+            this.headers = headers;
+        }
+
+        /**
+         * Returns headers, if any.  May be {@code null}.
+         */
+        public HttpHeaders headers() {
+            return headers;
+        }
+    }
 }

handler-proxy/src/test/java/io/netty/handler/proxy/HttpProxyHandlerTest.java

@@ -15,20 +15,37 @@
  */
 package io.netty.handler.proxy;
 
+import io.netty.bootstrap.Bootstrap;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelPromise;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.MultithreadEventLoopGroup;
+import io.netty.channel.local.LocalAddress;
+import io.netty.channel.local.LocalChannel;
+import io.netty.channel.local.LocalHandler;
+import io.netty.channel.local.LocalServerChannel;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
 import io.netty.handler.codec.http.DefaultHttpHeaders;
 import io.netty.handler.codec.http.FullHttpRequest;
 import io.netty.handler.codec.http.HttpHeaderNames;
 import io.netty.handler.codec.http.HttpHeaders;
+import io.netty.handler.codec.http.HttpResponseEncoder;
+import io.netty.handler.codec.http.HttpResponseStatus;
 import io.netty.handler.codec.http.HttpVersion;
+import io.netty.handler.proxy.HttpProxyHandler.HttpProxyConnectException;
 import io.netty.util.NetUtil;
+import java.util.concurrent.atomic.AtomicReference;
 import org.junit.Test;
 
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 
-import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.*;
 import static org.mockito.Mockito.*;
 
 public class HttpProxyHandlerTest {
@@ -153,6 +170,70 @@ public class HttpProxyHandlerTest {
                 true);
     }
 
+    @Test
+    public void testExceptionDuringConnect() throws Exception {
+        EventLoopGroup group = null;
+        Channel serverChannel = null;
+        Channel clientChannel = null;
+        try {
+            group = new MultithreadEventLoopGroup(1, LocalHandler.newFactory());
+            final LocalAddress addr = new LocalAddress("a");
+            final AtomicReference<Throwable> exception = new AtomicReference<Throwable>();
+            ChannelFuture sf =
+                new ServerBootstrap().channel(LocalServerChannel.class).group(group).childHandler(
+                    new ChannelInitializer<Channel>() {
+
+                        @Override
+                        protected void initChannel(Channel ch) {
+                            ch.pipeline().addFirst(new HttpResponseEncoder());
+                            ch.pipeline().addFirst(new ChannelInboundHandlerAdapter() {
+                                @Override
+                                public void channelActive(ChannelHandlerContext ctx) {
+                                    DefaultFullHttpResponse response = new DefaultFullHttpResponse(
+                                            HttpVersion.HTTP_1_1,
+                                            HttpResponseStatus.BAD_GATEWAY);
+                                    response.headers().add("name", "value");
+                                    response.headers().add(HttpHeaderNames.CONTENT_LENGTH, "0");
+                                    ctx.writeAndFlush(response);
+                                }
+                            });
+                        }
+                    }).bind(addr);
+            serverChannel = sf.sync().channel();
+            ChannelFuture cf = new Bootstrap().channel(LocalChannel.class).group(group).handler(
+                new ChannelInitializer<Channel>() {
+                    @Override
+                    protected void initChannel(Channel ch) {
+                        ch.pipeline().addFirst(new HttpProxyHandler(addr));
+                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
+                            @Override
+                            public void exceptionCaught(ChannelHandlerContext ctx,
+                                Throwable cause) {
+                                exception.set(cause);
+                            }
+                        });
+                    }
+                }).connect(new InetSocketAddress("localhost", 1234));
+            clientChannel = cf.sync().channel();
+            clientChannel.close().sync();
+
+            assertTrue(exception.get() instanceof HttpProxyConnectException);
+            HttpProxyConnectException actual = (HttpProxyConnectException) exception.get();
+            assertNotNull(actual.headers());
+            assertEquals("value", actual.headers().get("name"));
+        } finally {
+            if (clientChannel != null) {
+                clientChannel.close();
+            }
+            if (serverChannel != null) {
+                serverChannel.close();
+            }
+            if (group != null) {
+                group.shutdownGracefully();
+            }
+        }
+    }
+
     private static void testInitialMessage(InetSocketAddress socketAddress,
                                            String expectedUrl,
                                            String expectedHostHeader,