Generate an error for active connections when the HTTP client is shutdown

Before the underlying sockets where just silently closed causing no more IO events on active connections no matter what state they where in.

Also added an optional "force" argument to HttpClient.shutdown. If that is false (the default) The HttpClient will not close active connections until they are done. This causes all pkg/http and pub tests to pass.

R=ager@google.com, nweiz@google.com

BUG=dart:6594

Review URL: https://codereview.chromium.org//11411121

git-svn-id: https://dart.googlecode.com/svn/branches/bleeding_edge/dart@15275 260f80e4-7a28-3924-810f-c04153c831b5

pkg/http/test/request_test.dart

@@ -13,23 +13,16 @@ import 'utils.dart';
 
 void main() {
   test('.send', () {
-    print("This test is known to be flaky, please ignore "
-          "(debug prints below added by sgjesse@)");
-    print(".send test starting server...");
     startServer();
-    print(".send test server running");
 
     var request = new http.Request('POST', serverUrl);
     request.body = "hello";
     var future = request.send().chain((response) {
-      print(".send test response received");
       expect(response.statusCode, equals(200));
       return consumeInputStream(response.stream);
     }).transform((bytes) => new String.fromCharCodes(bytes));
     future.onComplete((_) {
-      print(".send test stopping server...");
       stopServer();
-      print(".send test server stopped");
     });
 
     expect(future, completion(parse(equals({
@@ -41,7 +34,6 @@ void main() {
       },
       'body': 'hello'
     }))));
-    print(".send test started");
   });
 
   group('#contentLength', () {

pkg/pkg.status

@@ -50,9 +50,6 @@ unittest/test/mock_regexp_negative_test: Fail
 unittest/test/mock_stepwise_negative_test: Fail
 args/test/args_test: Fail # http://dartbug.com/6790
 
-[ $system == windows ]
-http/test/request_test: Pass, Timeout # Issue 6594
-
 [ $compiler == dart2js || $compiler == dartc ]
 unittest/test/instance_test: Skip
 

sdk/lib/io/http.dart

@@ -830,9 +830,16 @@ abstract class HttpClient {
   set findProxy(String f(Uri url));
 
   /**
-   * Shutdown the HTTP client releasing all resources.
+   * Shutdown the HTTP client. If [force] is [:false:] (the default)
+   * the [:HttpClient:] will be kept alive until all active
+   * connections are done. If [force] is [:true:] any active
+   * connections will be closed to immediately release all
+   * resources. These closed connections will receive an [:onError:]
+   * callback to indicate that the client was shutdown. In both cases
+   * trying to establish a new connection after calling [shutdown]
+   * will throw an exception.
    */
-  void shutdown();
+  void shutdown({bool force: false});
 }
 
 

sdk/lib/io/http_impl.dart

@@ -1463,7 +1463,9 @@ class _HttpClientConnection
   }
 
   void _onError(e) {
-    // Socket is closed either due to an error or due to normal socket close.
+    if (_socketConn != null) {
+      _client._closeSocketConnection(_socketConn);
+    }
     if (_onErrorCallback != null) {
       _onErrorCallback(e);
     } else {
@@ -1473,9 +1475,6 @@ class _HttpClientConnection
     if (_response != null && _response._streamErrorHandler != null) {
        _response._streamErrorHandler(e);
     }
-    if (_socketConn != null) {
-      _client._closeSocketConnection(_socketConn);
-    }
   }
 
   void _onResponseReceived(int statusCode,
@@ -1490,11 +1489,17 @@ class _HttpClientConnection
   }
 
   void _onDataEnd(bool close) {
-    _response._onDataEnd();
     _state |= _HttpConnectionBase.RESPONSE_DONE;
+    _response._onDataEnd();
     _checkSocketDone();
   }
 
+  void _onClientShutdown() {
+    if (!_isResponseDone) {
+      _onError(new HttpException("Client shutdown"));
+    }
+  }
+
   void set onRequest(void handler(HttpClientRequest request)) {
     _onRequest = handler;
   }
@@ -1548,13 +1553,8 @@ class _HttpClientConnection
     var redirect = new _RedirectInfo(_response.statusCode, method, url);
     // The actual redirect is postponed until both response and
     // request are done.
-    if (_isAllDone) {
-      _doRedirect(redirect);
-    } else {
-      // Prepare for redirect.
-      assert(_pendingRetry == null);
-      _pendingRedirect = redirect;
-    }
+    assert(_pendingRetry == null);
+    _pendingRedirect = redirect;
   }
 
   List<RedirectInfo> get redirects => _redirects;
@@ -1594,12 +1594,14 @@ class _SocketConnection {
     _socket.onClosed = null;
     _socket.onError = null;
     _returnTime = new Date.now();
+    _httpClientConnection = null;
   }
 
   void _close() {
     _socket.onData = null;
     _socket.onClosed = null;
     _socket.onError = null;
+    _httpClientConnection = null;
     _socket.close();
   }
 
@@ -1611,6 +1613,7 @@ class _SocketConnection {
   int _port;
   Socket _socket;
   Date _returnTime;
+  HttpClientConnection _httpClientConnection;
 }
 
 class _ProxyConfiguration {
@@ -1730,19 +1733,22 @@ class _HttpClient implements HttpClient {
 
   set findProxy(String f(Uri uri)) => _findProxy = f;
 
-  void shutdown() {
-    _closeQueue.shutdown();
+  void shutdown({bool force: false}) {
+    if (force) _closeQueue.shutdown();
     _openSockets.forEach((String key, Queue<_SocketConnection> connections) {
       while (!connections.isEmpty) {
         _SocketConnection socketConn = connections.removeFirst();
         socketConn._socket.close();
       }
     });
-    _activeSockets.forEach((_SocketConnection socketConn) {
-      socketConn._socket.close();
-    });
+    if (force) {
+      _activeSockets.forEach((_SocketConnection socketConn) {
+        socketConn._socket.close();
+        socketConn._httpClientConnection._onClientShutdown();
+      });
+    }
     if (_evictionTimer != null) _cancelEvictionTimer();
-     _shutdown = true;
+    _shutdown = true;
   }
 
   void _cancelEvictionTimer() {
@@ -1768,6 +1774,7 @@ class _HttpClient implements HttpClient {
       void _connectionOpened(_SocketConnection socketConn,
                              _HttpClientConnection connection,
                              bool usingProxy) {
+        socketConn._httpClientConnection = connection;
         connection._usingProxy = usingProxy;
         connection._connectionEstablished(socketConn);
         HttpClientRequest request = connection.open(method, url);
@@ -1882,15 +1889,15 @@ class _HttpClient implements HttpClient {
   }
 
   void _returnSocketConnection(_SocketConnection socketConn) {
-    // Mark socket as returned to unregister from the old connection.
-    socketConn._markReturned();
-
-    // If the HTTP client is beeing shutdown don't return the connection.
+    // If the HTTP client is being shutdown don't return the connection.
     if (_shutdown) {
-      socketConn._socket.close();
+      socketConn._close();
       return;
     };
 
+    // Mark socket as returned to unregister from the old connection.
+    socketConn._markReturned();
+
     String key = _connectionKey(socketConn._host, socketConn._port);
 
     // Get or create the connection list for this key.

sdk/lib/io/http_parser.dart

@@ -582,6 +582,8 @@ class _HttpParser {
   }
 
   void streamDone() {
+    String type() => _requestParser ? "request" : "response";
+
     // If the connection is idle the HTTP stream is closed.
     if (_state == _State.START) {
       if (_requestParser) {
@@ -589,7 +591,7 @@ class _HttpParser {
       } else {
         error(
             new HttpParserException(
-                "Connection closed before full header was received"));
+                "Connection closed before full ${type()} header was received"));
       }
       return;
     }
@@ -600,7 +602,7 @@ class _HttpParser {
       // throw the error.
       error(
           new HttpParserException(
-              "Connection closed before full header was received"));
+                "Connection closed before full ${type()} header was received"));
       return;
     }
 
@@ -614,7 +616,7 @@ class _HttpParser {
       // throw the error.
       error(
           new HttpParserException(
-              "Connection closed before full body was received"));
+                "Connection closed before full ${type()} body was received"));
     }
   }
 

tests/standalone/io/http_advanced_test.dart

@@ -287,9 +287,12 @@ Future testHost() {
     };
     conn.onResponse = (HttpClientResponse response) {
       Expect.equals(HttpStatus.OK, response.statusCode);
-      httpClient.shutdown();
-      testServerMain.shutdown();
-      completer.complete(true);
+      response.inputStream.onData = response.inputStream.read;
+      response.inputStream.onClosed = () {
+        httpClient.shutdown();
+        testServerMain.shutdown();
+        completer.complete(true);
+      };
     };
   });
   testServerMain.start();
@@ -309,12 +312,15 @@ Future testExpires() {
                     response.headers["expires"][0]);
       Expect.equals(new Date.utc(1999, Date.JUN, 11, 18, 46, 53, 0),
                     response.headers.expires);
-      responses++;
-      if (responses == 2) {
-        httpClient.shutdown();
-        testServerMain.shutdown();
-        completer.complete(true);
-      }
+      response.inputStream.onData = response.inputStream.read;
+      response.inputStream.onClosed = () {
+        responses++;
+        if (responses == 2) {
+          httpClient.shutdown();
+          testServerMain.shutdown();
+          completer.complete(true);
+        }
+      };
     }
 
     HttpClientConnection conn1 = httpClient.get("127.0.0.1", port, "/expires1");
@@ -346,12 +352,15 @@ Future testContentType() {
       Expect.equals("html", response.headers.contentType.subType);
       Expect.equals("utf-8",
                     response.headers.contentType.parameters["charset"]);
-      responses++;
-      if (responses == 2) {
-        httpClient.shutdown();
-        testServerMain.shutdown();
-        completer.complete(true);
-      }
+      response.inputStream.onData = response.inputStream.read;
+      response.inputStream.onClosed = () {
+        responses++;
+        if (responses == 2) {
+          httpClient.shutdown();
+          testServerMain.shutdown();
+          completer.complete(true);
+        }
+      };
     }
 
     HttpClientConnection conn1 =
@@ -415,10 +424,13 @@ Future testCookies() {
         request.cookies.add(response.cookies[1]);
         request.outputStream.close();
       };
-      conn2.onResponse = (HttpClientResponse ignored) {
-        httpClient.shutdown();
-        testServerMain.shutdown();
-        completer.complete(true);
+      conn2.onResponse = (HttpClientResponse response) {
+        response.inputStream.onData = response.inputStream.read;
+        response.inputStream.onClosed = () {
+          httpClient.shutdown();
+          testServerMain.shutdown();
+          completer.complete(true);
+        };
       };
     };
   });
@@ -439,9 +451,12 @@ Future testFlush() {
     };
     conn.onResponse = (HttpClientResponse response) {
       Expect.equals(HttpStatus.OK, response.statusCode);
-      httpClient.shutdown();
-      testServerMain.shutdown();
-      completer.complete(true);
+      response.inputStream.onData = response.inputStream.read;
+      response.inputStream.onClosed = () {
+        httpClient.shutdown();
+        testServerMain.shutdown();
+        completer.complete(true);
+      };
     };
   });
   testServerMain.start();

tests/standalone/io/http_auth_test.dart

@@ -80,8 +80,11 @@ void testUrlUserInfo() {
           new Uri.fromString(
               "http://username:password@127.0.0.1:${server.port}/"));
   conn.onResponse = (HttpClientResponse response) {
-    server.shutdown();
-    client.shutdown();
+    response.inputStream.onData = response.inputStream.read;
+    response.inputStream.onClosed = () {
+      server.shutdown();
+      client.shutdown();
+    };
   };
 }
 
@@ -94,7 +97,8 @@ void testBasicNoCredentials() {
     HttpClientConnection conn = client.getUrl(url);
     conn.onResponse = (HttpClientResponse response) {
       Expect.equals(HttpStatus.UNAUTHORIZED, response.statusCode);
-      completer.complete(null);
+      response.inputStream.onData = response.inputStream.read;
+      response.inputStream.onClosed = () => completer.complete(null);
     };
     return completer.future;
   }
@@ -123,7 +127,8 @@ void testBasicCredentials() {
     HttpClientConnection conn = client.getUrl(url);
     conn.onResponse = (HttpClientResponse response) {
       Expect.equals(HttpStatus.OK, response.statusCode);
-      completer.complete(null);
+      response.inputStream.onData = response.inputStream.read;
+      response.inputStream.onClosed = () => completer.complete(null);
     };
     return completer.future;
   }
@@ -175,7 +180,8 @@ void testBasicAuthenticateCallback() {
     HttpClientConnection conn = client.getUrl(url);
     conn.onResponse = (HttpClientResponse response) {
       Expect.equals(HttpStatus.OK, response.statusCode);
-      completer.complete(null);
+      response.inputStream.onData = response.inputStream.read;
+      response.inputStream.onClosed = () => completer.complete(null);
     };
     return completer.future;
   }

tests/standalone/io/http_connection_close_test.dart

@@ -66,10 +66,11 @@ void testStreamResponse() {
     resp.inputStream.onData = () {
       bytes += resp.inputStream.read().length;
       if (bytes > 100) {
-        client.shutdown();
+        client.shutdown(force: true);
       }
     };
   };
+  connection.onError = (e) => Expect.isTrue(e is HttpException);
 }
 
 main() {

tests/standalone/io/http_connection_header_test.dart

@@ -64,11 +64,13 @@ void test(int totalConnections, bool clientPersistentConnection) {
       Expect.isFalse(response.persistentConnection);
       checkExpectedConnectionHeaders(response.headers,
                                      response.persistentConnection);
-      count++;
-      if (count == totalConnections) {
-        client.shutdown();
-        server.close();
-      }
+      response.inputStream.onClosed = () {
+        count++;
+        if (count == totalConnections) {
+          client.shutdown();
+          server.close();
+        }
+      };
     };
   }
 }

tests/standalone/io/http_content_length_test.dart

@@ -38,10 +38,13 @@ void testNoBody(int totalConnections, bool explicitContentLength) {
       Expect.equals("0", response.headers.value('content-length'));
       Expect.equals(0, response.contentLength);
       count++;
-      if (count == totalConnections) {
-        client.shutdown();
-        server.close();
-      }
+      response.inputStream.onData = response.inputStream.read;
+      response.inputStream.onClosed = () {
+        if (count == totalConnections) {
+          client.shutdown();
+          server.close();
+        }
+      };
     };
   }
 }
@@ -82,10 +85,13 @@ void testBody(int totalConnections) {
       Expect.equals("2", response.headers.value('content-length'));
       Expect.equals(2, response.contentLength);
       count++;
-      if (count == totalConnections) {
-        client.shutdown();
-        server.close();
-      }
+      response.inputStream.onData = response.inputStream.read;
+      response.inputStream.onClosed = () {
+        if (count == totalConnections) {
+          client.shutdown();
+          server.close();
+        }
+      };
     };
   }
 }

tests/standalone/io/http_redirect_test.dart

@@ -91,7 +91,7 @@ void testManualRedirect() {
      client.getUrl(new Uri.fromString("http://127.0.0.1:${server.port}/1"));
   conn.followRedirects = false;
   conn.onResponse = (HttpClientResponse response) {
-    response.inputStream.onData = () => response.inputStream.read();
+    response.inputStream.onData = response.inputStream.read;
     response.inputStream.onClosed = () {
       redirectCount++;
       if (redirectCount < 10) {

tests/standalone/io/http_server_early_client_close_test.dart

@@ -63,17 +63,18 @@ void testEarlyClose() {
   // The empty packet is valid.
 
   // Close while sending header
-  add("G", "Connection closed before full header was received");
-  add("GET /", "Connection closed before full header was received");
-  add("GET / HTTP/1.1", "Connection closed before full header was received");
-  add("GET / HTTP/1.1\r\n", "Connection closed before full header was received");
+  String message = "Connection closed before full request header was received";
+  add("G", message);
+  add("GET /", message);
+  add("GET / HTTP/1.1", message);
+  add("GET / HTTP/1.1\r\n", message);
 
   // Close while sending content
   add("GET / HTTP/1.1\r\nContent-Length: 100\r\n\r\n",
-      "Connection closed before full body was received",
+      "Connection closed before full request body was received",
       expectRequest: true);
   add("GET / HTTP/1.1\r\nContent-Length: 100\r\n\r\n1",
-      "Connection closed before full body was received",
+      "Connection closed before full request body was received",
       expectRequest: true);
 
 

tests/standalone/io/http_server_handler_test.dart

@@ -101,7 +101,7 @@ void testDefaultHandler() {
       conn.onResponse = (HttpClientResponse response) {
         Expect.equals(HttpStatus.NOT_FOUND, response.statusCode);
         Expect.equals("Non Trouvé", response.reasonPhrase);
-        done();
+        response.inputStream.onClosed = done;
       };
       conn.onError = error;
     };

tests/standalone/io/http_session_test.dart

@@ -30,8 +30,11 @@ Future<String> connectGetSession(int port, [String session]) {
     request.outputStream.close();
   };
   conn.onResponse = (response) {
-    client.shutdown();
-    c.complete(getSessionId(response.cookies));
+    response.inputStream.onData = response.inputStream.read;
+    response.inputStream.onClosed = () {
+      client.shutdown();
+      c.complete(getSessionId(response.cookies));
+    };
   };
   return c.future;
 }

tests/standalone/io/http_shutdown_test.dart

@@ -22,11 +22,13 @@ void test1(int totalConnections) {
       request.outputStream.close();
     };
     conn.onResponse = (HttpClientResponse response) {
-      count++;
-      if (count == totalConnections) {
-        client.shutdown();
-        server.close();
-      }
+      response.inputStream.onClosed = () {
+        count++;
+        if (count == totalConnections) {
+          client.shutdown();
+          server.close();
+        }
+      };
     };
   }
 }
@@ -51,11 +53,14 @@ void test2(int totalConnections) {
       request.outputStream.close();
     };
     conn.onResponse = (HttpClientResponse response) {
-      count++;
-      if (count == totalConnections) {
-        client.shutdown();
-        server.close();
-      }
+      response.inputStream.onData = response.inputStream.read;
+      response.inputStream.onClosed = () {
+        count++;
+        if (count == totalConnections) {
+          client.shutdown();
+          server.close();
+        }
+      };
     };
   }
 }
@@ -85,11 +90,14 @@ void test3(int totalConnections) {
       request.outputStream.close();
     };
     conn.onResponse = (HttpClientResponse response) {
-      count++;
-      if (count == totalConnections) {
-        client.shutdown();
-        server.close();
-      }
+      response.inputStream.onData = response.inputStream.read;
+      response.inputStream.onClosed = () {
+        count++;
+        if (count == totalConnections) {
+          client.shutdown();
+          server.close();
+        }
+      };
     };
   }
 }
@@ -113,6 +121,7 @@ void test4() {
   var client= new HttpClient();
   var conn = client.get("127.0.0.1", server.port, "/");
   conn.onResponse = (var response) {
+    response.inputStream.onData = response.inputStream.read;
     response.inputStream.onClosed = () {
       client.shutdown();
     };
@@ -137,13 +146,14 @@ void test5(int totalConnections) {
   for (int i = 0; i < totalConnections; i++) {
     var conn = client.post("127.0.0.1", server.port, "/");
     conn.onRequest = (req) { req.outputStream.write([0]); };
+    conn.onError = (e) => Expect.isTrue(e is HttpException);
   }
   bool clientClosed = false;
   new Timer.repeating(100, (timer) {
     if (!clientClosed) {
       if (server.connectionsInfo().total == totalConnections) {
         clientClosed = true;
-        client.shutdown();
+        client.shutdown(force: true);
       }
     } else {
       if (server.connectionsInfo().total == 0) {