Don't close existing connection on HttpServer close (and stream cancel).

sdk/lib/io/http_impl.dart

 // Copyright (c) 2013, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 part of dart.io;
 
 class _HttpIncoming extends Stream<List<int>> {
   final int _transferLength;
   final Completer _dataCompleter = new Completer();
   Stream<List<int>> _stream;
@@ skipping 520 matching lines @@
     // TODO(ajohnsen): Currently, contentLength, chunkedTransferEncoding and
     // persistentConnection is not guaranteed to be in sync.
     if (!_headersWritten) {
       if (!_ignoreBody && headers.contentLength == -1) {
         // If no body was written, _ignoreBody is false (it's not a HEAD
         // request) and the content-length is unspecified, set contentLength to
         // 0.
         headers.chunkedTransferEncoding = false;
         headers.contentLength = 0;
       } else if (!_ignoreBody && headers.contentLength > 0) {
-        _headersSink.close().catchError((_) {});
-        return new Future.error(new HttpException(
+        _headersSink.addError(new HttpException(
             "No content while contentLength was specified to be greater "
-            " than 0: ${headers.contentLength}.",
+            "than 0: ${headers.contentLength}.",
             uri: _uri));
+        return _headersSink.done;
       }
     }
     return _writeHeaders().then((_) => _headersSink.close());
   }
 
   void _writeHeader();  // TODO(ajohnsen): Better name.
 }
 
 
 class _HttpOutboundConsumer implements StreamConsumer {
@@ skipping 57 matching lines @@
 
   Future addStream(var stream) {
     // If we saw a socket error subscribe and then cancel, to ignore any data
     // on the stream.
     if (_socketError) {
       stream.listen(null).cancel();
       return new Future.value(_outbound);
     }
     _completer = new Completer();
     _subscription = stream.listen(
-        (data) {
-          _controller.add(data);
-        },
+        (data) => _controller.add(data),
         onDone: _done,
-        onError: _done,
+        onError: (e, s) => _controller.addError(e, s),
         cancelOnError: true);
     // Pause the first request.
     if (_controller == null) _subscription.pause();
     _ensureController();
     return _completer.future;
   }
 
   Future close() {
     Future closeOutbound() {
       if (_socketError) return new Future.value(_outbound);
@@ skipping 1233 matching lines @@
                                            incoming.headers.protocolVersion,
                                            outgoing,
                                            _httpServer.serverHeader);
           var request = new _HttpRequest(response, incoming, _httpServer, this);
           _streamFuture = outgoing.done
               .then((_) {
                 response.deadline = null;
                 if (_state == _DETACHED) return;
                 if (response.persistentConnection &&
                     request.persistentConnection &&
-                    incoming.fullBodyRead) {
+                    incoming.fullBodyRead &&
+                    !_httpParser.upgrade &&
+                    !_httpServer.closed) {
                   _state = _IDLE;
                   _startTimeout();
                   // Resume the subscription for incoming requests as the
                   // request is now processed.
                   _subscription.resume();
                 } else {
                   // Close socket, keep-alive not used or body sent before
                   // received data was handled.
                   destroy();
                 }
@@ skipping 108 matching lines @@
           _connections.add(connection);
         },
         onError: _controller.addError,
         onDone: _controller.close);
     return _controller.stream.listen(onData,
                                      onError: onError,
                                      onDone: onDone,
                                      cancelOnError: cancelOnError);
   }
 
-  Future close() {
+  Future close({bool force: false}) {
     closed = true;
     Future result;
     if (_serverSocket != null && _closeServer) {
       result = _serverSocket.close();
     } else {
       result = new Future.value();
     }
-    if (_sessionManagerInstance != null) {
+    if (force) {
+      for (var c in _connections.toList()) {
+        c.destroy();
+      }
+      assert(_connections.isEmpty);
+    } else {
+      for (var c in _connections.where((c) => c._isIdle).toList()) {
+        c.destroy();
+      }
+    }
+    _maybeCloseSessionManager();
+    return result;
+  }
+
+  void _maybeCloseSessionManager() {
+    if (closed &&
+        _connections.isEmpty &&
+        _sessionManagerInstance != null) {
       _sessionManagerInstance.close();
       _sessionManagerInstance = null;
     }
-    for (_HttpConnection connection in _connections.toList()) {
-      connection.destroy();
-    }
-    _connections.clear();
-    return result;
   }
 
   int get port {
     if (closed) throw new HttpException("HttpServer is not bound to a socket");
     return _serverSocket.port;
   }
 
   InternetAddress get address {
     if (closed) throw new HttpException("HttpServer is not bound to a socket");
     return _serverSocket.address;
   }
 
   set sessionTimeout(int timeout) {
     _sessionManager.sessionTimeout = timeout;
   }
 
   void _handleRequest(HttpRequest request) {
     _controller.add(request);
   }
 
   void _handleError(error) {
     if (!closed) _controller.addError(error);
   }
 
   void _connectionClosed(_HttpConnection connection) {
     _connections.remove(connection);
+    _maybeCloseSessionManager();
   }
 
   _HttpSessionManager get _sessionManager {
     // Lazy init.
     if (_sessionManagerInstance == null) {
       _sessionManagerInstance = new _HttpSessionManager();
     }
     return _sessionManagerInstance;
   }
 
@@ skipping 106 matching lines @@
   final String username;
   final String password;
   final bool isDirect;
 }
 
 
 class _HttpConnectionInfo implements HttpConnectionInfo {
   static _HttpConnectionInfo create(Socket socket) {
     if (socket == null) return null;
     try {
-      _HttpConnectionInfo info = new _HttpConnectionInfo._();
+      _HttpConnectionInfo info = new _HttpConnectionInfo();
       info.remoteHost = socket.remoteHost;
       info.remotePort = socket.remotePort;
       info.localPort = socket.port;
       return info;
     } catch (e) { }
     return null;
   }
 
-  _HttpConnectionInfo._();
-
   String remoteHost;
   int remotePort;
   int localPort;
 }
 
 
 class _DetachedSocket extends Stream<List<int>> implements Socket {
   final Stream<List<int>> _incoming;
   final Socket _socket;
 
@@ skipping 294 matching lines @@
   final Uri location;
 }
 
 String _getHttpVersion() {
   var version = Platform.version;
   // Only include major and minor version numbers.
   int index = version.indexOf('.', version.indexOf('.') + 1);
   version = version.substring(0, index);
   return 'Dart/$version (dart:io)';
 }