Change StreamController constructor.

sdk/lib/io/http_parser.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;
 
 // Global constants.
 class _Const {
   // Bytes for "HTTP".
   static const HTTP = const [72, 84, 84, 80];
@@ skipping 88 matching lines @@
 
   List<int> carryOverData;
   bool paused;
 
   Completer resumeCompleter;
 
   _HttpDetachedIncoming(StreamSubscription this.subscription,
                         List<int> this.carryOverData,
                         Completer oldResumeCompleter) {
     controller = new StreamController<List<int>>(
-        onSubscriptionStateChange: onSubscriptionStateChange,
-        onPauseStateChange: onPauseStateChange);
+        onListen: resume,
+        onPause: pause,
+        onResume: resume,
+        onCancel: () => subscription.cancel());
     if (subscription == null) {
       // Socket was already closed.
       if (carryOverData != null) controller.add(carryOverData);
       controller.close();
     } else {
       pause();
       if (oldResumeCompleter != null) oldResumeCompleter.complete();
       subscription.resume();
       subscription.onData(controller.add);
       subscription.onDone(controller.close);
@@ skipping 28 matching lines @@
     }
   }
 
   void pause() {
     paused = true;
     if (resumeCompleter == null) {
       resumeCompleter = new Completer();
       subscription.pause(resumeCompleter.future);
     }
   }
-
-  void onPauseStateChange() {
-    if (controller.isPaused) {
-      pause();
-    } else {
-      resume();
-    }
-  }
-
-  void onSubscriptionStateChange() {
-    if (controller.hasListener) {
-      resume();
-    } else {
-      subscription.cancel();
-    }
-  }
 }
 
 
 /**
  * HTTP parser which parses the data stream given to [consume].
  *
  * If an HTTP parser error occours, the parser will signal an error to either
  * the current _HttpIncoming or the _parser itself.
  *
  * The connection upgrades (e.g. switching from HTTP/1.1 to the
@@ skipping 17 matching lines @@
   factory _HttpParser.requestParser() {
     return new _HttpParser._(true);
   }
 
   factory _HttpParser.responseParser() {
     return new _HttpParser._(false);
   }
 
   _HttpParser._(this._requestParser) {
     _controller = new StreamController<_HttpIncoming>(
-          onSubscriptionStateChange: _updateParsePauseState,
-          onPauseStateChange: _updateParsePauseState);
+          onListen: _updateParsePauseState,
+          onPause: _updateParsePauseState,
+          onResume: _updateParsePauseState,
+          onCancel: _updateParsePauseState);
     _reset();
   }
 
 
   StreamSubscription<_HttpIncoming> listen(void onData(_HttpIncoming event),
                                            {void onError(AsyncError error),
                                             void onDone(),
                                             bool unsubscribeOnError}) {
     return _controller.stream.listen(onData,
                                      onError: onError,
@@ skipping 651 matching lines @@
       return byte - 0x61 + 10;  // a - f
     } else {
       throw new HttpParserException("Failed to parse HTTP");
     }
   }
 
   void _createIncoming(int transferLength) {
     assert(_incoming == null);
     assert(_bodyController == null);
     _bodyController = new StreamController<List<int>>(
-        onSubscriptionStateChange: _bodySubscriptionStateChange,
-        onPauseStateChange: _updateParsePauseState);
+        onListen: _bodySubscriptionStateChange,
+        onPause: _updateParsePauseState,
+        onResume: _updateParsePauseState,
+        onCancel: _bodySubscriptionStateChange);
     _incoming = new _HttpIncoming(
         _headers, transferLength, _bodyController.stream);
     _pauseParsing();  // Needed to handle detaching - don't start on the body!
   }
 
   void _closeIncoming() {
     // Ignore multiple close (can happend in re-entrance).
     if (_incoming == null) return;
     var tmp = _incoming;
     _incoming = null;
@@ skipping 82 matching lines @@
   StreamController<_HttpIncoming> _controller;
   StreamController<List<int>> _bodyController;
 }
 
 
 class HttpParserException implements Exception {
   const HttpParserException([String this.message = ""]);
   String toString() => "HttpParserException: $message";
   final String message;
 }