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.

3 files changed, 44 insertions, 26 deletions

diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpConnection.java b/jetty-client/src/main/java/org/eclipse/jetty/client/HttpConnection.java
index d578f1cc85..dfd11c25b0 100644
--- a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpConnection.java
+++ b/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
-    public void fillInterested()
+    protected boolean isClosed()
     {
-        // This is necessary when "upgrading" the connection for example after proxied
-        // 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();
+        return closed;
     }
 
     @Override
diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpDestination.java b/jetty-client/src/main/java/org/eclipse/jetty/client/HttpDestination.java
index 94744d0ed1..9b2caf889d 100644
--- a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpDestination.java
+++ b/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());
-                        connection.close();
+                        tunnelFailed(connection, delegate, result.getFailure());
                     }
                     else
                     {
                         Response response = result.getResponse();
                         if (response.getStatus() == 200)
                         {
-                            // Wrap the connection with TLS
-                            Connection tunnel = client.tunnel(connection);
-                            delegate.succeeded(tunnel);
+                            tunnelSucceeded(connection, delegate);
                         }
                         else
                         {
-                            failed(new HttpResponseException("Received " + response + " for " + result.getRequest(), response));
-                            connection.close();
+                            tunnelFailed(connection, delegate, new HttpResponseException("Received " + response + " for " + result.getRequest(), response));
                         }
                     }
                 }
diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpReceiver.java b/jetty-client/src/main/java/org/eclipse/jetty/client/HttpReceiver.java
index b3f8d51eac..8bb6c6d3dc 100644
--- a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpReceiver.java
+++ b/jetty-client/src/main/java/org/eclipse/jetty/client/HttpReceiver.java
@@ -68,21 +68,30 @@ public class HttpReceiver implements HttpParser.ResponseHandler<ByteBuffer>
         {
             while (true)
             {
-                int read = endPoint.fill(buffer);
-                LOG.debug("Read {} bytes from {}", read, connection);
-                if (read > 0)
+                // Connection may be closed in a parser callback
+                if (connection.isClosed())
                 {
-                    parse(buffer);
-                }
-                else if (read == 0)
-                {
-                    fillInterested();
+                    LOG.debug("{} closed", connection);
                     break;
                 }
                 else
                 {
-                    shutdown();
-                    break;
+                    int read = endPoint.fill(buffer);
+                    LOG.debug("Read {} bytes from {}", read, connection);
+                    if (read > 0)
+                    {
+                        parse(buffer);
+                    }
+                    else if (read == 0)
+                    {
+                        fillInterested();
+                        break;
+                    }
+                    else
+                    {
+                        shutdown();
+                        break;
+                    }
                 }
             }
         }
@@ -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