432145 - Pending request is not failed when HttpClient is stopped.

Fixed by fixing the code in close() to also abort pending exchanges.
Reviewed for HTTP, FastCGI and SPDY transports.

4 files changed, 76 insertions, 10 deletions

diff --git a/jetty-spdy/spdy-http-client-transport/src/main/java/org/eclipse/jetty/spdy/client/http/HttpChannelOverSPDY.java b/jetty-spdy/spdy-http-client-transport/src/main/java/org/eclipse/jetty/spdy/client/http/HttpChannelOverSPDY.java
index bd19978803..00f87cd498 100644
--- a/jetty-spdy/spdy-http-client-transport/src/main/java/org/eclipse/jetty/spdy/client/http/HttpChannelOverSPDY.java
+++ b/jetty-spdy/spdy-http-client-transport/src/main/java/org/eclipse/jetty/spdy/client/http/HttpChannelOverSPDY.java
@@ -21,17 +21,20 @@ package org.eclipse.jetty.spdy.client.http;
 import org.eclipse.jetty.client.HttpChannel;
 import org.eclipse.jetty.client.HttpDestination;
 import org.eclipse.jetty.client.HttpExchange;
+import org.eclipse.jetty.client.api.Result;
 import org.eclipse.jetty.spdy.api.Session;
 
 public class HttpChannelOverSPDY extends HttpChannel
 {
+    private final HttpConnectionOverSPDY connection;
     private final Session session;
     private final HttpSenderOverSPDY sender;
     private final HttpReceiverOverSPDY receiver;
 
-    public HttpChannelOverSPDY(HttpDestination destination, Session session)
+    public HttpChannelOverSPDY(HttpDestination destination, HttpConnectionOverSPDY connection, Session session)
     {
         super(destination);
+        this.connection = connection;
         this.session = session;
         this.sender = new HttpSenderOverSPDY(this);
         this.receiver = new HttpReceiverOverSPDY(this);
@@ -72,4 +75,11 @@ public class HttpChannelOverSPDY extends HttpChannel
         sender.abort(cause);
         return receiver.abort(cause);
     }
+
+    @Override
+    public void exchangeTerminated(Result result)
+    {
+        super.exchangeTerminated(result);
+        connection.release(this);
+    }
 }
diff --git a/jetty-spdy/spdy-http-client-transport/src/main/java/org/eclipse/jetty/spdy/client/http/HttpConnectionOverSPDY.java b/jetty-spdy/spdy-http-client-transport/src/main/java/org/eclipse/jetty/spdy/client/http/HttpConnectionOverSPDY.java
index ef96e4158e..57ba0652e2 100644
--- a/jetty-spdy/spdy-http-client-transport/src/main/java/org/eclipse/jetty/spdy/client/http/HttpConnectionOverSPDY.java
+++ b/jetty-spdy/spdy-http-client-transport/src/main/java/org/eclipse/jetty/spdy/client/http/HttpConnectionOverSPDY.java
@@ -18,6 +18,9 @@
 
 package org.eclipse.jetty.spdy.client.http;
 
+import java.nio.channels.AsynchronousCloseException;
+import java.util.Set;
+
 import org.eclipse.jetty.client.HttpChannel;
 import org.eclipse.jetty.client.HttpConnection;
 import org.eclipse.jetty.client.HttpDestination;
@@ -25,9 +28,11 @@ import org.eclipse.jetty.client.HttpExchange;
 import org.eclipse.jetty.spdy.api.GoAwayInfo;
 import org.eclipse.jetty.spdy.api.Session;
 import org.eclipse.jetty.util.Callback;
+import org.eclipse.jetty.util.ConcurrentHashSet;
 
 public class HttpConnectionOverSPDY extends HttpConnection
 {
+    private final Set<HttpChannel> channels = new ConcurrentHashSet<>();
     private final Session session;
 
     public HttpConnectionOverSPDY(HttpDestination destination, Session session)
@@ -41,14 +46,35 @@ public class HttpConnectionOverSPDY extends HttpConnection
     {
         normalizeRequest(exchange.getRequest());
         // One connection maps to N channels, so for each exchange we create a new channel
-        HttpChannel channel = new HttpChannelOverSPDY(getHttpDestination(), session);
+        HttpChannel channel = new HttpChannelOverSPDY(getHttpDestination(), this, session);
+        channels.add(channel);
         channel.associate(exchange);
         channel.send();
     }
 
+    protected void release(HttpChannel channel)
+    {
+        channels.remove(channel);
+    }
+
     @Override
     public void close()
     {
+        // First close then abort, to be sure that the connection cannot be reused
+        // from an onFailure() handler or by blocking code waiting for completion.
+        getHttpDestination().close(this);
         session.goAway(new GoAwayInfo(), new Callback.Adapter());
+        abort(new AsynchronousCloseException());
+    }
+
+    private void abort(Throwable failure)
+    {
+        for (HttpChannel channel : channels)
+        {
+            HttpExchange exchange = channel.getHttpExchange();
+            if (exchange != null)
+                exchange.getRequest().abort(failure);
+        }
+        channels.clear();
     }
 }
diff --git a/jetty-spdy/spdy-http-client-transport/src/main/java/org/eclipse/jetty/spdy/client/http/HttpDestinationOverSPDY.java b/jetty-spdy/spdy-http-client-transport/src/main/java/org/eclipse/jetty/spdy/client/http/HttpDestinationOverSPDY.java
index dff5a0785b..bdc4c05b5f 100644
--- a/jetty-spdy/spdy-http-client-transport/src/main/java/org/eclipse/jetty/spdy/client/http/HttpDestinationOverSPDY.java
+++ b/jetty-spdy/spdy-http-client-transport/src/main/java/org/eclipse/jetty/spdy/client/http/HttpDestinationOverSPDY.java
@@ -35,12 +35,4 @@ public class HttpDestinationOverSPDY extends MultiplexHttpDestination<HttpConnec
     {
         connection.send(exchange);
     }
-
-    @Override
-    public void abort(Throwable cause)
-    {
-        // TODO: in case of connection failure, we need to abort also
-        // TODO: all pending exchanges, so we need to track them.
-        super.abort(cause);
-    }
 }
diff --git a/jetty-spdy/spdy-http-client-transport/src/test/java/org/eclipse/jetty/spdy/client/http/HttpClientTest.java b/jetty-spdy/spdy-http-client-transport/src/test/java/org/eclipse/jetty/spdy/client/http/HttpClientTest.java
index 0de7a209ab..4fa701746d 100644
--- a/jetty-spdy/spdy-http-client-transport/src/test/java/org/eclipse/jetty/spdy/client/http/HttpClientTest.java
+++ b/jetty-spdy/spdy-http-client-transport/src/test/java/org/eclipse/jetty/spdy/client/http/HttpClientTest.java
@@ -23,6 +23,7 @@ import java.net.URI;
 import java.net.URLEncoder;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -36,6 +37,7 @@ import javax.servlet.http.HttpServletResponse;
 import org.eclipse.jetty.client.api.ContentResponse;
 import org.eclipse.jetty.client.api.Request;
 import org.eclipse.jetty.client.api.Response;
+import org.eclipse.jetty.client.api.Result;
 import org.eclipse.jetty.client.util.BytesContentProvider;
 import org.eclipse.jetty.http.HttpMethod;
 import org.eclipse.jetty.server.handler.AbstractHandler;
@@ -426,4 +428,40 @@ public class HttpClientTest extends AbstractHttpClientServerTest
         Assert.assertEquals(200, response.getStatus());
         Assert.assertEquals(length, response.getContent().length);
     }
+
+    @Test
+    public void testLongPollIsAbortedWhenClientIsStopped() throws Exception
+    {
+        final CountDownLatch latch = new CountDownLatch(1);
+        start(new AbstractHandler()
+        {
+            @Override
+            public void handle(String target, org.eclipse.jetty.server.Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException
+            {
+                baseRequest.setHandled(true);
+                request.startAsync();
+                latch.countDown();
+            }
+        });
+
+        final CountDownLatch completeLatch = new CountDownLatch(1);
+        client.newRequest("localhost", connector.getLocalPort())
+                .scheme(scheme)
+                .send(new Response.CompleteListener()
+                {
+                    @Override
+                    public void onComplete(Result result)
+                    {
+                        if (result.isFailed())
+                            completeLatch.countDown();
+                    }
+                });
+
+        Assert.assertTrue(latch.await(5, TimeUnit.SECONDS));
+
+        // Stop the client, the complete listener must be invoked.
+        client.stop();
+
+        Assert.assertTrue(completeLatch.await(5, TimeUnit.SECONDS));
+    }
 }