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) => _controller.addError(e),

[Søren Gjesse, 2013/10/23 08:14:05]

Do we want to propagate the stack trace here?

[Anders Johnsen, 2013/10/23 10:23:24]

Yes, nice catch! :)

         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) {
                   _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();
                 }
               })
               .catchError((e) {
                 destroy();
               });
           response._ignoreBody = request.method == "HEAD";
           response._httpRequest = request;
-          _httpServer._handleRequest(request);
+          _httpServer._handleRequest(request, this);
         },
         onDone: () {
           destroy();
         },
         onError: (error) {
           // Ignore failed requests that was closed before headers was received.
           destroy();
         });
   }
 
@@ skipping 91 matching lines @@
           _connections.add(connection);
         },
         onError: _controller.addError,
         onDone: _controller.close);
     return _controller.stream.listen(onData,
                                      onError: onError,
                                      onDone: onDone,
                                      cancelOnError: cancelOnError);
   }
 
   Future close() {

[Søren Gjesse, 2013/10/23 08:14:05]

I think we still need the public close to do a full shutdown closing all client
connections. We could add an optional boolean force argument, but as far as I
can see that would be (almost) the same as canceling the subscription.

Currently close has a future which completes when the server socket is closed.
There is no future for when all the active connections are all done. If (when)
StreamSubscription.cancel gets a future when should that complete?

Alternatively we could have a destroy method which closes all active
connections. The future from close can then complete when all active connections
are terminated.

[Anders Johnsen, 2013/10/23 10:23:24]

The 'problem' is that there is no binding between a server-socket and the
accepted sockets. Even if the server-socket is closed, the accepted sockets will
live on. The question is then, what should a HTTP server do with the live
requests. What we did before was to destroy them, not caring about what they
were doing. That's a very aggressive approach, as the user may get broken data,
depending on when it's closed.

Moving this into a 'destroy' method makes sense. However, as it's exactly the
same semantics, I would suggest we do as with HttpClient and add a `force` flag,
with `false` as default. I think users explicit should express when they are
okay with breaking requests. What do you think?

[Søren Gjesse, 2013/10/23 11:05:28]

I am fine with adding the force flag. In case force is false should the returned
future then complete when all client connections are done or should we have a
separate future for that?

[Anders Johnsen, 2013/10/23 12:06:55]

The way we force-close is by calling 'socket.destroy'. This means that it's a
sync operation with nothing to wait for, so it'll just be the same future in
both cases.

     closed = true;
     Future result;
     if (_serverSocket != null && _closeServer) {
       result = _serverSocket.close();
     } else {
       result = new Future.value();
     }
-    if (_sessionManagerInstance != null) {
+    _maybeCloseSessionManager();
+    return result;
+  }
+
+  void _maybeCloseSessionManager() {
+    if (closed &&
+        _connections.length == 0 &&
+        _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) {
+  void _handleRequest(HttpRequest request, _HttpConnection connection) {
+    if (closed) {
+      connection.destroy();
+      return;
+    }
     _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)';
 }