Add 'deadline' to HttpResponse, making it possible to give a max duration for the liveness of the r…

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 668 matching lines @@
   }
 }
 
 
 class _HttpResponse extends _HttpOutboundMessage<HttpResponse>
     implements HttpResponse {
   int statusCode = 200;
   String _reasonPhrase;
   List<Cookie> _cookies;
   _HttpRequest _httpRequest;
+  Duration _deadline;
+  Timer _deadlineTimer;
 
   _HttpResponse(Uri uri,
                 String protocolVersion,
-                _HttpOutgoing _outgoing,
+                _HttpOutgoing outgoing,
                 String serverHeader)
-      : super(uri, protocolVersion, _outgoing) {
+      : super(uri, protocolVersion, outgoing) {
     if (serverHeader != null) headers.set('Server', serverHeader);
   }
 
   List<Cookie> get cookies {
     if (_cookies == null) _cookies = new List<Cookie>();
     return _cookies;
   }
 
   String get reasonPhrase => _findReasonPhrase(statusCode);
   void set reasonPhrase(String reasonPhrase) {
     if (_headersWritten) throw new StateError("Header already sent");
     _reasonPhrase = reasonPhrase;
   }
 
   Future<Socket> detachSocket() {
     if (_headersWritten) throw new StateError("Headers already sent");
+    deadline = null;  // Be sure to stop any deadline.
     var future = _httpRequest._httpConnection.detachSocket();
     _writeHeaders(drainRequest: false).then((_) => close());
     // Close connection so the socket is 'free'.
     close();
     done.catchError((_) {
       // Catch any error on done, as they automatically will be
       // propagated to the websocket.
     });
     return future;
   }
 
   HttpConnectionInfo get connectionInfo => _httpRequest.connectionInfo;
 
+  Duration get deadline => _deadline;
+
+  void set deadline(Duration d) {
+    if (_deadlineTimer != null) _deadlineTimer.cancel();
+    _deadline = d;
+
+    if (_deadline == null) return;
+    _deadlineTimer = new Timer(_deadline, () {
+      _outgoing.socket.destroy();
+    });
+  }
+
   void _writeHeader() {
     var builder = new BytesBuilder();
     writeSP() => builder.add(const [_CharCode.SP]);
     writeCRLF() => builder.add(const [_CharCode.CR, _CharCode.LF]);
 
     // Write status line.
     if (headers.protocolVersion == "1.1") {
       builder.add(_Const.HTTP11);
     } else {
       builder.add(_Const.HTTP10);
@@ skipping 357 matching lines @@
           uri: uri));
     }
     sink.close();
   }
 }
 
 
 // Extends StreamConsumer as this is an internal type, only used to pipe to.
 class _HttpOutgoing implements StreamConsumer<List<int>> {
   final Completer _doneCompleter = new Completer();
-  final StreamConsumer _consumer;
+  final Socket socket;
 
-  _HttpOutgoing(StreamConsumer this._consumer);
+  _HttpOutgoing(Socket this.socket);
 
   Future addStream(Stream<List<int>> stream) {
-    return _consumer.addStream(stream)
+    return socket.addStream(stream)
         .catchError((error) {
           _doneCompleter.completeError(error);
           throw error;
         });
   }
 
   Future close() {
-    _doneCompleter.complete(_consumer);
+    _doneCompleter.complete(socket);
     return new Future.value();
   }
 
   Future get done => _doneCompleter.future;
 }
 
 class _HttpClientConnection {
   final String key;
   final Socket _socket;
   final bool _proxyTunnel;
@@ skipping 680 matching lines @@
           _subscription.pause();
           _state = _ACTIVE;
           var outgoing = new _HttpOutgoing(_socket);
           var response = new _HttpResponse(incoming.uri,
                                            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) {
                   _state = _IDLE;
                   // 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
@@ skipping 583 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)';
 }