[master] [grizzly-provider] connection abort, when neither transfer-encoding nor content-length is specified

providers/grizzly/src/main/java/org/asynchttpclient/providers/grizzly/EventHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2013 Sonatype, Inc. All rights reserved.
+ * Copyright (c) 2013-2014 Sonatype, Inc. All rights reserved.
  *
  * This program is licensed to you under the Apache License Version 2.0,
  * and you may not use this file except in compliance with the Apache License Version 2.0.
@@ -163,8 +163,9 @@ public void onInitialLineParsed(HttpHeader httpHeader, FilterChainContext ctx) {
                 }
             }
         }
-        final GrizzlyResponseStatus responseStatus = new GrizzlyResponseStatus((HttpResponsePacket) httpHeader, context.getRequest()
-                .getURI(), config);
+        final GrizzlyResponseStatus responseStatus =
+                new GrizzlyResponseStatus((HttpResponsePacket) httpHeader,
+                        context.getRequest().getURI(), config);
         context.setResponseStatus(responseStatus);
         if (context.getStatusHandler() != null) {
             return;

providers/grizzly/src/main/java/org/asynchttpclient/providers/grizzly/GrizzlyAsyncHttpProvider.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2012 Sonatype, Inc. All rights reserved.
+ * Copyright (c) 2012-2014 Sonatype, Inc. All rights reserved.
  *
  * This program is licensed to you under the Apache License Version 2.0,
  * and you may not use this file except in compliance with the Apache License Version 2.0.
@@ -73,6 +73,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import org.glassfish.grizzly.spdy.SpdyVersion;
 
 /**
  * A Grizzly 2.0-based implementation of {@link AsyncHttpProvider}.
@@ -441,8 +442,9 @@ void timeout(final Connection c) {
     // ---------------------------------------------------------- Nested Classes
 
     private static final class ProtocolNegotiator implements ClientSideNegotiator {
+        private static final SpdyVersion[] SUPPORTED_SPDY_VERSIONS =
+                {SpdyVersion.SPDY_3_1, SpdyVersion.SPDY_3};
 
-        private static final String SPDY = "spdy/3";
         private static final String HTTP = "HTTP/1.1";
 
         private final FilterChain spdyFilterChain;
@@ -465,23 +467,31 @@ public boolean wantNegotiate(SSLEngine engine) {
         }
 
         @Override
-        public String selectProtocol(SSLEngine engine, LinkedHashSet<String> strings) {
-            GrizzlyAsyncHttpProvider.LOGGER.info("ProtocolSelector::selectProtocol: " + strings);
+        public String selectProtocol(SSLEngine engine, LinkedHashSet<String> protocols) {
+            GrizzlyAsyncHttpProvider.LOGGER.info("ProtocolSelector::selectProtocol: " + protocols);
             final Connection connection = NextProtoNegSupport.getConnection(engine);
 
-            // Give preference to SPDY/3.  If not available, check for HTTP as a
-            // fallback
-            if (strings.contains(SPDY)) {
-                GrizzlyAsyncHttpProvider.LOGGER.info("ProtocolSelector::selecting: " + SPDY);
-                SSLConnectionContext sslCtx = SSLUtils.getSslConnectionContext(connection);
-                sslCtx.setNewConnectionFilterChain(spdyFilterChain);
-                final SpdySession spdySession = new SpdySession(connection, false, spdyHandlerFilter);
-                spdySession.setLocalInitialWindowSize(spdyHandlerFilter.getInitialWindowSize());
-                spdySession.setLocalMaxConcurrentStreams(spdyHandlerFilter.getMaxConcurrentStreams());
-                Utils.setSpdyConnection(connection);
-                SpdySession.bind(connection, spdySession);
-                return SPDY;
-            } else if (strings.contains(HTTP)) {
+            // Give preference to SPDY/3.1 or SPDY/3.  If not available, check for HTTP as a
+            // fallback            
+            for (SpdyVersion version : SUPPORTED_SPDY_VERSIONS) {
+                final String versionDef = version.toString();
+                if (protocols.contains(versionDef)) {
+                    GrizzlyAsyncHttpProvider.LOGGER.info("ProtocolSelector::selecting: " + versionDef);
+                    SSLConnectionContext sslCtx = SSLUtils.getSslConnectionContext(connection);
+                    sslCtx.setNewConnectionFilterChain(spdyFilterChain);
+                    final SpdySession spdySession =
+                            version.newSession(connection, false, spdyHandlerFilter);
+
+                    spdySession.setLocalStreamWindowSize(spdyHandlerFilter.getInitialWindowSize());
+                    spdySession.setLocalMaxConcurrentStreams(spdyHandlerFilter.getMaxConcurrentStreams());
+                    Utils.setSpdyConnection(connection);
+                    SpdySession.bind(connection, spdySession);
+
+                    return versionDef;
+                }
+            }
+
+            if (protocols.contains(HTTP)) {
                 GrizzlyAsyncHttpProvider.LOGGER.info("ProtocolSelector::selecting: " + HTTP);
                 // Use the default HTTP FilterChain.
                 return HTTP;

providers/grizzly/src/main/java/org/asynchttpclient/providers/grizzly/GrizzlyResponseStatus.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2012 Sonatype, Inc. All rights reserved.
+ * Copyright (c) 2012-2014 Sonatype, Inc. All rights reserved.
  *
  * This program is licensed to you under the Apache License Version 2.0,
  * and you may not use this file except in compliance with the Apache License Version 2.0.
@@ -38,7 +38,8 @@ public class GrizzlyResponseStatus extends HttpResponseStatus {
     private final int majorVersion;
     private final int minorVersion;
     private final String protocolText;
-
+    private final HttpResponsePacket response;
+
     // ------------------------------------------------------------ Constructors
 
     public GrizzlyResponseStatus(final HttpResponsePacket response, final URI uri, AsyncHttpClientConfig config) {
@@ -49,6 +50,8 @@ public GrizzlyResponseStatus(final HttpResponsePacket response, final URI uri, A
         majorVersion = response.getProtocol().getMajorVersion();
         minorVersion = response.getProtocol().getMinorVersion();
         protocolText = response.getProtocolString();
+
+        this.response = response;
     }
 
     // ----------------------------------------- Methods from HttpResponseStatus
@@ -105,4 +108,11 @@ public int getProtocolMinorVersion() {
     public String getProtocolText() {
         return protocolText;
     }
+
+    /**
+     * @return internal Grizzly {@link HttpResponsePacket}
+     */
+    public HttpResponsePacket getResponse() {
+        return response;
+    }
 }

providers/grizzly/src/main/java/org/asynchttpclient/providers/grizzly/HttpTxContext.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2013 Sonatype, Inc. All rights reserved.
+ * Copyright (c) 2013-2014 Sonatype, Inc. All rights reserved.
  *
  * This program is licensed to you under the Apache License Version 2.0,
  * and you may not use this file except in compliance with the Apache License Version 2.0.
@@ -33,6 +33,10 @@
 import java.io.IOException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import org.asynchttpclient.providers.grizzly.filters.events.GracefulCloseEvent;
+import org.glassfish.grizzly.Connection;
+import org.glassfish.grizzly.filterchain.FilterChain;
+import org.glassfish.grizzly.http.HttpResponsePacket;
 
 public final class HttpTxContext {
 
@@ -62,10 +66,21 @@ public final class HttpTxContext {
     private HandShake handshake;
     private ProtocolHandler protocolHandler;
     private WebSocket webSocket;
-    private CloseListener listener = new CloseListener<Closeable, CloseType>() {
+    private final CloseListener listener = new CloseListener<Closeable, CloseType>() {
         @Override
         public void onClosed(Closeable closeable, CloseType type) throws IOException {
-            if (CloseType.REMOTELY.equals(type)) {
+            if (isGracefullyFinishResponseOnClose()) {
+                // Connection was closed.
+                // This event is fired only for responses, which don't have
+                // associated transfer-encoding or content-length.
+                // We have to complete such a request-response processing gracefully.
+                final Connection c = responseStatus.getResponse()
+                        .getRequest().getConnection();
+                final FilterChain fc = (FilterChain) c.getProcessor();
+
+                fc.fireEventUpstream(c,
+                        new GracefulCloseEvent(HttpTxContext.this), null);
+            } else if (CloseType.REMOTELY.equals(type)) {
                 abort(AsyncHttpProviderUtils.REMOTELY_CLOSED_EXCEPTION);
             }
         }
@@ -252,6 +267,12 @@ public void setWebSocket(WebSocket webSocket) {
         this.webSocket = webSocket;
     }
 
+    private boolean isGracefullyFinishResponseOnClose() {
+        final HttpResponsePacket response = responseStatus.getResponse();
+        return !response.getProcessingState().isKeepAlive() &&
+                !response.isChunked() && response.getContentLength() == -1;
+    }
+
     // ------------------------------------------------- Package Private Methods
 
     public HttpTxContext copy() {

providers/grizzly/src/main/java/org/asynchttpclient/providers/grizzly/filters/AsyncHttpClientEventFilter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2013 Sonatype, Inc. All rights reserved.
+ * Copyright (c) 2013-2014 Sonatype, Inc. All rights reserved.
  *
  * This program is licensed to you under the Apache License Version 2.0,
  * and you may not use this file except in compliance with the Apache License Version 2.0.
@@ -21,6 +21,10 @@
 import org.glassfish.grizzly.http.HttpHeader;
 
 import java.io.IOException;
+import org.asynchttpclient.providers.grizzly.filters.events.GracefulCloseEvent;
+import org.glassfish.grizzly.filterchain.FilterChainEvent;
+import org.glassfish.grizzly.filterchain.NextAction;
+import org.glassfish.grizzly.http.HttpResponsePacket;
 
 /**
  * Extension of the {@link HttpClientFilter} that is responsible for handling
@@ -45,6 +49,28 @@ public AsyncHttpClientEventFilter(final EventHandler eventHandler, final int max
         this.eventHandler = eventHandler;
     }
 
+    @Override
+    public NextAction handleEvent(final FilterChainContext ctx,
+            final FilterChainEvent event) throws IOException {
+        if (event.type() == GracefulCloseEvent.class) {
+            // Connection was closed.
+            // This event is fired only for responses, which don't have
+            // associated transfer-encoding or content-length.
+            // We have to complete such a request-response processing gracefully.
+            final GracefulCloseEvent closeEvent = (GracefulCloseEvent) event;
+            final HttpResponsePacket response = closeEvent.getHttpTxContext()
+                    .getResponseStatus().getResponse();
+            response.getProcessingState().getHttpContext().attach(ctx);
+
+            onHttpPacketParsed(response, ctx);
+
+            return ctx.getStopAction();
+        }
+
+        return ctx.getInvokeAction();
+    }
+
+
     @Override
     public void exceptionOccurred(FilterChainContext ctx, Throwable error) {
         eventHandler.exceptionOccurred(ctx, error);

providers/grizzly/src/main/java/org/asynchttpclient/providers/grizzly/filters/AsyncHttpClientFilter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2013 Sonatype, Inc. All rights reserved.
+ * Copyright (c) 2013-2014 Sonatype, Inc. All rights reserved.
  *
  * This program is licensed to you under the Apache License Version 2.0,
  * and you may not use this file except in compliance with the Apache License Version 2.0.
@@ -281,21 +281,26 @@ private boolean sendAsGrizzlyRequest(final RequestInfoHolder requestInfoHolder,
             sendingCtx = checkAndHandleFilterChainUpdate(ctx, sendingCtx);
         }
         final Connection c = ctx.getConnection();
+        final HttpContext httpCtx;
         if (!Utils.isSpdyConnection(c)) {
-            HttpContext.newInstance(ctx, c, c, c);
+            httpCtx = HttpContext.newInstance(c, c, c, requestPacketLocal);
         } else {
             SpdySession session = SpdySession.get(c);
             final Lock lock = session.getNewClientStreamLock();
             try {
                 lock.lock();
                 SpdyStream stream = session.openStream(requestPacketLocal, session.getNextLocalStreamId(), 0, 0, 0, false,
                         !requestPacketLocal.isExpectContent());
-                HttpContext.newInstance(ctx, stream, stream, stream);
+                httpCtx = HttpContext.newInstance(stream, stream, stream, requestPacketLocal);
             } finally {
                 lock.unlock();
             }
         }
+        httpCtx.attach(ctx);
         HttpTxContext.set(ctx, httpTxContext);
+        requestPacketLocal.getProcessingState().setHttpContext(httpCtx);
+        requestPacketLocal.setConnection(c);
+
         return sendRequest(sendingCtx, request, requestPacketLocal);
     }
 

providers/grizzly/src/main/java/org/asynchttpclient/providers/grizzly/filters/events/GracefulCloseEvent.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright (c) 2014 Sonatype, Inc. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+ */
+
+package org.asynchttpclient.providers.grizzly.filters.events;
+
+import org.asynchttpclient.providers.grizzly.HttpTxContext;
+import org.glassfish.grizzly.filterchain.FilterChainEvent;
+
+/**
+ * {@link FilterChainEvent} to gracefully complete the request-response processing
+ * when {@link Connection} is getting closed by the remote host.
+ *
+ * @since 1.8.7
+ * @author The Grizzly Team
+ */
+public class GracefulCloseEvent implements FilterChainEvent {
+    private final HttpTxContext httpTxContext;
+
+    public GracefulCloseEvent(HttpTxContext httpTxContext) {
+        this.httpTxContext = httpTxContext;
+    }
+
+    public HttpTxContext getHttpTxContext() {
+        return httpTxContext;
+    }
+
+    @Override
+    public Object type() {
+        return GracefulCloseEvent.class;
+    }
+}