Optmize detached socket to not contain a extra stream controller/subscription, but simly forward ca…

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 78 matching lines @@
   static const int HTTP11 = 2;
 }
 
 // States of the HTTP parser state machine.
 class _MessageType {
   static const int UNDETERMINED = 0;
   static const int REQUEST = 1;
   static const int RESPONSE = 0;
 }
 
-class _HttpDetachedIncoming extends Stream<List<int>> {
-  StreamController<List<int>> controller;
-  final StreamSubscription subscription;
 
-  List<int> bufferedData;
-  bool paused;
+/**
+ * The _HttpDetachedStreamSubscription takes a subscription and some extra data,
+ * and makes it possible to "inject" the data in from of other data events
+ * from the subscription.
+ *
+ * It does so by overriding pause/resume, so that once the
+ * _HttpDetachedStreamSubscription is resumed, it'll deliver the data before
+ * resuming the underlaying subscription.
+ */
+class _HttpDetachedStreamSubscription implements StreamSubscription<List<int>> {
+  StreamSubscription<List<int>> _subscription;
+  List<int> _injectData;
+  bool _isCanceled = false;
+  int _pauseCount = 1;
+  Function _userOnData;
+  bool _scheduled = false;
 
-  Completer resumeCompleter;
+  _HttpDetachedStreamSubscription(this._subscription,
+                                  this._injectData,
+                                  this._userOnData);
 
-  _HttpDetachedIncoming(this.subscription, this.bufferedData) {
-    controller = new StreamController<List<int>>(
-        sync: true,
-        onListen: resume,
-        onPause: pause,
-        onResume: resume,
-        onCancel: () => subscription.cancel());
-    if (subscription == null) {
-      // Socket was already closed.
-      if (bufferedData != null) controller.add(bufferedData);
-      controller.close();
+  bool get isPaused => _subscription.isPaused;
+
+  Future asFuture([futureValue]) => _subscription.asFuture(futureValue);
+
+  Future cancel() {
+    _isCanceled = true;
+    _injectData = null;
+    return _subscription.cancel();
+  }
+
+  void onData(void handleData(List<int> data)) {
+    _userOnData = handleData;
+    _subscription.onData(handleData);
+  }
+
+  void onDone(void handleDone()) {
+    _subscription.onDone(handleDone);
+  }
+
+  void onError(Function handleError) {
+    _subscription.onError(handleError);
+  }
+
+  void pause([Future resumeSignal]) {
+    if (_injectData == null) {
+      _subscription.pause(resumeSignal);
     } else {
-      pause();
-      subscription
-          ..resume()
-          ..onData(controller.add)
-          ..onDone(controller.close)
-          ..onError(controller.addError);
+      _pauseCount++;
+      if (resumeSignal != null) {
+        resumeSignal.whenComplete(resume);
+      }
     }
   }
 
+  void resume() {
+    if (_injectData == null) {
+      _subscription.resume();
+    } else {
+      _pauseCount--;
+      _maybeScheduleData();
+    }
+  }
+
+  void _maybeScheduleData() {
+    if (_scheduled) return;
+    if (_pauseCount != 0) return;
+    _scheduled = true;
+    scheduleMicrotask(() {
+      _scheduled = false;
+      if (_pauseCount > 0 || _isCanceled) return;
+      var data = _injectData;
+      _injectData = null;
+      // To ensure that 'subscription.isPaused' is false, we resume the
+      // subscription here. This is fine as potential events are delayed.
+      _subscription.resume();
+      if (_userOnData != null) {
+        _userOnData(data);
+      }
+    });
+  }
+}
+
+
+class _HttpDetachedIncoming extends Stream<List<int>> {
+  final StreamSubscription subscription;
+  final List<int> bufferedData;
+
+  _HttpDetachedIncoming(this.subscription, this.bufferedData);
+
   StreamSubscription<List<int>> listen(void onData(List<int> event),
                                        {Function onError,
                                         void onDone(),
                                         bool cancelOnError}) {
-    return controller.stream.listen(
-        onData,
-        onError: onError,
-        onDone: onDone,
-        cancelOnError: cancelOnError);
-  }
-
-  void resume() {
-    paused = false;
-    if (bufferedData != null) {
-      var data = bufferedData;
-      bufferedData = null;
-      controller.add(data);
-      // If the consumer pauses again after the carry-over data, we'll not
-      // continue our subscriber until the next resume.
-      if (paused) return;
-    }
-    if (resumeCompleter != null) {
-      resumeCompleter.complete();
-      resumeCompleter = null;
-    }
-  }
-
-  void pause() {
-    paused = true;
-    if (resumeCompleter == null) {
-      resumeCompleter = new Completer();
-      subscription.pause(resumeCompleter.future);
+    if (subscription != null) {
+      subscription
+        ..onData(onData)
+        ..onError(onError)
+        ..onDone(onDone);
+      if (bufferedData == null) {
+        return subscription..resume();
+      }
+      return new _HttpDetachedStreamSubscription(subscription,
+                                                 bufferedData,
+                                                 onData)
+        ..resume();
+    } else {
+      return new Stream.fromIterable(bufferedData)
+          .listen(onData,
+                  onError: onError,
+                  onDone: onDone,
+                  cancelOnError: cancelOnError);
     }
   }
 }
 
 
 /**
  * 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.
@@ skipping 814 matching lines @@
     }
   }
 
   void _reportError(error, [stackTrace]) {
     if (_socketSubscription != null) _socketSubscription.cancel();
     _state = _State.FAILURE;
     _controller.addError(error, stackTrace);
     _controller.close();
   }
 }