Fix behavior when listening multiple times to a broadcast take/skip stream.

sdk/lib/async/stream_pipe.dart

Index: sdk/lib/async/stream_pipe.dart
diff --git a/sdk/lib/async/stream_pipe.dart b/sdk/lib/async/stream_pipe.dart
index ca1164dea81bb084211ee36eaff5e128bd744228..6b67fd6232bfb2420353c0a049bf60e9fccf8683 100644
--- a/sdk/lib/async/stream_pipe.dart
+++ b/sdk/lib/async/stream_pipe.dart
@@ -308,20 +308,32 @@ class _HandleErrorStream<T> extends _ForwardingStream<T, T> {
 
 
 class _TakeStream<T> extends _ForwardingStream<T, T> {
-  int _remaining;
+  final int _count;
 
   _TakeStream(Stream<T> source, int count)
-      : this._remaining = count, super(source) {
+      : this._count = count, super(source) {
     // This test is done early to avoid handling an async error
     // in the _handleData method.
     if (count is! int) throw new ArgumentError(count);
   }
 
+  StreamSubscription<T> _createSubscription(
+      void onData(T data),
+      Function onError,
+      void onDone(),
+      bool cancelOnError) {
+    return new _StateStreamSubscription<T>(
+        this, onData, onError, onDone, cancelOnError, _count);
+  }
+
   void _handleData(T inputEvent, _EventSink<T> sink) {
-    if (_remaining > 0) {
+    _StateStreamSubscription subscription = sink;
+    int count = subscription._count;
+    if (count > 0) {
       sink._add(inputEvent);
-      _remaining -= 1;
-      if (_remaining == 0) {
+      count -= 1;
+      subscription._count = count;
+      if (count == 0) {
         // Closing also unsubscribes all subscribers, which unsubscribes
         // this from source.
         sink._close();
@@ -330,6 +342,26 @@ class _TakeStream<T> extends _ForwardingStream<T, T> {
   }
 }
 
+/**
+ * A [_ForwardingStreamSubscription] with one extra state field.
+ *
+ * Use by several different classes, some storeing an integer, others a bool.

[floitsch, 2015/02/13 13:52:56]

storing

+ */
+class _StateStreamSubscription<T> extends _ForwardingStreamSubscription<T, T> {
+  // Raw state field. Typed access provided by getters and setters below.
+  var _sharedState;
+
+  _StateStreamSubscription(_ForwardingStream stream, void onData(T data),
+                           Function onError, void onDone(),
+                           bool cancelOnError, this._sharedState)
+      : super(stream, onData, onError, onDone, cancelOnError);
+
+  bool get _flag => _sharedState;
+  void set _flag(bool flag) { _sharedState = flag; }
+  int get _count => _sharedState;
+  void set _count(int count) { _sharedState = count; }
+}
+
 
 class _TakeWhileStream<T> extends _ForwardingStream<T, T> {
   final _Predicate<T> _test;
@@ -356,18 +388,29 @@ class _TakeWhileStream<T> extends _ForwardingStream<T, T> {
 }
 
 class _SkipStream<T> extends _ForwardingStream<T, T> {
-  int _remaining;
+  final int _count;
 
   _SkipStream(Stream<T> source, int count)
-      : this._remaining = count, super(source) {
+      : this._count = count, super(source) {
     // This test is done early to avoid handling an async error
     // in the _handleData method.
     if (count is! int || count < 0) throw new ArgumentError(count);
   }
 
+  StreamSubscription<T> _createSubscription(
+      void onData(T data),
+      Function onError,
+      void onDone(),
+      bool cancelOnError) {
+    return new _StateStreamSubscription<T>(
+        this, onData, onError, onDone, cancelOnError, _count);
+  }
+
   void _handleData(T inputEvent, _EventSink<T> sink) {
-    if (_remaining > 0) {
-      _remaining--;
+    _StateStreamSubscription subscription = sink;
+    int count = subscription._count;
+    if (count > 0) {
+      subscription._count = count - 1;
       return;
     }
     sink._add(inputEvent);
@@ -376,13 +419,23 @@ class _SkipStream<T> extends _ForwardingStream<T, T> {
 
 class _SkipWhileStream<T> extends _ForwardingStream<T, T> {
   final _Predicate<T> _test;
-  bool _hasFailed = false;
 
   _SkipWhileStream(Stream<T> source, bool test(T value))
       : this._test = test, super(source);
 
+  StreamSubscription<T> _createSubscription(
+      void onData(T data),
+      Function onError,
+      void onDone(),
+      bool cancelOnError) {
+    return new _StateStreamSubscription<T>(
+        this, onData, onError, onDone, cancelOnError, false);
+  }
+
   void _handleData(T inputEvent, _EventSink<T> sink) {
-    if (_hasFailed) {
+    _StateStreamSubscription subscription = sink;
+    bool hasFailed = subscription._flag;
+    if (hasFailed) {
       sink._add(inputEvent);
       return;
     }
@@ -392,11 +445,11 @@ class _SkipWhileStream<T> extends _ForwardingStream<T, T> {
     } catch (e, s) {
       _addErrorWithReplacement(sink, e, s);
       // A failure to return a boolean is considered "not matching".
-      _hasFailed = true;
+      subscription._flag = true;
       return;
     }
     if (!satisfies) {
-      _hasFailed = true;
+      subscription._flag = true;
       sink._add(inputEvent);
     }
   }