414972 - HttpClient may read bytes with pre-tunnelled connection.

Now the receiver checks whether the connection is closed, and returns
immediately if it is without "stealing" the bytes to the tunnelled
connection.

jetty-client/src/main/java/org/eclipse/jetty/client/HttpConnection.java

@@ -53,7 +53,7 @@ public class HttpConnection extends AbstractConnection implements Connection
     private final HttpSender sender;
     private final HttpReceiver receiver;
     private long idleTimeout;
-    private volatile boolean closed;
+    private boolean closed;
 
     public HttpConnection(HttpClient client, EndPoint endPoint, HttpDestination destination)
     {
@@ -88,15 +88,9 @@ public class HttpConnection extends AbstractConnection implements Connection
         super.onClose();
     }
 
-    @Override
+    protected boolean isClosed()
-    public void fillInterested()
     {
-        // This is necessary when "upgrading" the connection for example after proxied
+        return closed;
-        // CONNECT requests, because the old connection will read the CONNECT response
-        // and then set the read interest, while the new connection attached to the same
-        // EndPoint also will set the read interest, causing a ReadPendingException.
-        if (!closed)
-            super.fillInterested();
     }
 
     @Override

jetty-client/src/main/java/org/eclipse/jetty/client/HttpDestination.java

@@ -428,6 +428,19 @@ public class HttpDestination implements Destination, Closeable, Dumpable
         getResponseNotifier().notifyComplete(listeners, new Result(request, cause, response, cause));
     }
 
+    protected void tunnelSucceeded(Connection connection, Promise<Connection> promise)
+    {
+        // Wrap the connection with TLS
+        Connection tunnel = client.tunnel(connection);
+        promise.succeeded(tunnel);
+    }
+
+    protected void tunnelFailed(Connection connection, Promise<Connection> promise, Throwable failure)
+    {
+        promise.failed(failure);
+        connection.close();
+    }
+
     @Override
     public String dump()
     {
@@ -516,22 +529,18 @@ public class HttpDestination implements Destination, Closeable, Dumpable
                 {
                     if (result.isFailed())
                     {
-                        failed(result.getFailure());
+                        tunnelFailed(connection, delegate, result.getFailure());
-                        connection.close();
                     }
                     else
                     {
                         Response response = result.getResponse();
                         if (response.getStatus() == 200)
                         {
-                            // Wrap the connection with TLS
+                            tunnelSucceeded(connection, delegate);
-                            Connection tunnel = client.tunnel(connection);
-                            delegate.succeeded(tunnel);
                         }
                         else
                         {
-                            failed(new HttpResponseException("Received " + response + " for " + result.getRequest(), response));
+                            tunnelFailed(connection, delegate, new HttpResponseException("Received " + response + " for " + result.getRequest(), response));
-                            connection.close();
                         }
                     }
                 }

jetty-client/src/main/java/org/eclipse/jetty/client/HttpReceiver.java

@@ -67,6 +67,14 @@ public class HttpReceiver implements HttpParser.ResponseHandler<ByteBuffer>
         try
         {
             while (true)
+            {
+                // Connection may be closed in a parser callback
+                if (connection.isClosed())
+                {
+                    LOG.debug("{} closed", connection);
+                    break;
+                }
+                else
                 {
                     int read = endPoint.fill(buffer);
                     LOG.debug("Read {} bytes from {}", read, connection);
@@ -86,6 +94,7 @@ public class HttpReceiver implements HttpParser.ResponseHandler<ByteBuffer>
                     }
                 }
             }
+        }
         catch (EofException x)
         {
             LOG.ignore(x);
@@ -415,6 +424,12 @@ public class HttpReceiver implements HttpParser.ResponseHandler<ByteBuffer>
         return updated;
     }
 
+    @Override
+    public String toString()
+    {
+        return String.format("%s@%x on %s", getClass().getSimpleName(), hashCode(), connection);
+    }
+
     private enum State
     {
         IDLE, RECEIVE, FAILURE