Make StreamSubscription be the active part of a stream.

sdk/lib/async/stream_pipe.dart

 // Copyright (c) 2012, 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.async;
 
 /**
  * Utility function to attach a stack trace to an [error]  if it doesn't have
  * one already.
  */
@@ skipping 51 matching lines @@
   StreamSubscription<T> _createSubscription(void onData(T value),
                                             void onError(error),
                                             void onDone(),
                                             bool cancelOnError) {
     return new _ForwardingStreamSubscription<S, T>(
         this, onData, onError, onDone, cancelOnError);
   }
 
   // Override the following methods in subclasses to change the behavior.
 
-  void _handleData(S data, _EventOutputSink<T> sink) {
+  void _handleData(S data, _EventSink<T> sink) {
     var outputData = data;
-    sink._sendData(outputData);
+    sink._add(outputData);
   }
 
-  void _handleError(error, _EventOutputSink<T> sink) {
-    sink._sendError(error);
+  void _handleError(error, _EventSink<T> sink) {
+    sink._addError(error);
   }
 
-  void _handleDone(_EventOutputSink<T> sink) {
-    sink._sendDone();
+  void _handleDone(_EventSink<T> sink) {
+    sink._close();
   }
 }
 
 /**
- * Common behavior of [StreamSubscription] classes.
- *
- * Stores and allows updating of the event handlers of a [StreamSubscription].
- */
-abstract class _BaseStreamSubscription<T> implements StreamSubscription<T> {
-  // TODO(ahe): Restore type when feature is implemented in dart2js
-  // checked mode. http://dartbug.com/7733
-  var /* _DataHandler<T> */ _onData;
-  _ErrorHandler _onError;
-  _DoneHandler _onDone;
-
-  _BaseStreamSubscription(this._onData,
-                          this._onError,
-                          this._onDone) {
-    if (_onData == null) _onData = _nullDataHandler;
-    if (_onError == null) _onError = _nullErrorHandler;
-    if (_onDone == null) _onDone = _nullDoneHandler;
-  }
-
-  // StreamSubscription interface.
-  void onData(void handleData(T event)) {
-    if (handleData == null) handleData = _nullDataHandler;
-    _onData = handleData;
-  }
-
-  void onError(void handleError(error)) {
-    if (handleError == null) handleError = _nullErrorHandler;
-    _onError = handleError;
-  }
-
-  void onDone(void handleDone()) {
-    if (handleDone == null) handleDone = _nullDoneHandler;
-    _onDone = handleDone;
-  }
-
-  void pause([Future resumeSignal]);
-
-  void resume();
-
-  void cancel();
-
-  Future asFuture([var futureValue]) {
-    _FutureImpl<T> result = new _FutureImpl<T>();
-
-    // Overwrite the onDone and onError handlers.
-    onDone(() { result._setValue(futureValue); });
-    onError((error) {
-      cancel();
-      result._setError(error);
-    });
-
-    return result;
-  }
-}
-
-
-/**
  * Abstract superclass for subscriptions that forward to other subscriptions.
  */
 class _ForwardingStreamSubscription<S, T>
-    extends _BaseStreamSubscription<T> implements _EventOutputSink<T> {
+    extends _BufferingStreamSubscription<T> {
   final _ForwardingStream<S, T> _stream;
-  final bool _cancelOnError;
 
   StreamSubscription<S> _subscription;
 
   _ForwardingStreamSubscription(this._stream,
                                 void onData(T data),
                                 void onError(error),
                                 void onDone(),
-                                this._cancelOnError)
-      : super(onData, onError, onDone) {
-    // Don't unsubscribe on incoming error, only if we send an error forwards.
+                                bool cancelOnError)
+      : super(onData, onError, onDone, cancelOnError) {
     _subscription =
         _stream._source.listen(_handleData,
                                onError: _handleError,
                                onDone: _handleDone);
   }
 
-  // StreamSubscription interface.
+  // _StreamSink interface.
+  // Transformers sending more than one event have no way to know if the stream
+  // is canceled or closed after the first, so we just ignore remaining events.
 
-  void pause([Future resumeSignal]) {
-    if (_subscription == null) return;
-    _subscription.pause(resumeSignal);
+  void _add(T data) {
+    if (_isClosed) return;
+    super._add(data);
   }
 
-  void resume() {
+  void _addError(Object error) {
+    if (_isClosed) return;
+    super._addError(error);
+  }
+
+  // StreamSubscription callbacks.
+
+  void _onPause() {
+    if (_subscription == null) return;
+    _subscription.pause();
+  }
+
+  void _onResume() {
     if (_subscription == null) return;
     _subscription.resume();
   }
 
-  bool get isPaused {
-    if (_subscription == null) return false;
-    return _subscription.isPaused;
-  }
-
-  void cancel() {
+  void _onCancel() {
     if (_subscription != null) {
-      _subscription.cancel();
+      StreamSubscription subscription = _subscription;
       _subscription = null;
+      subscription.cancel();
     }
   }
 
-  // _EventOutputSink interface. Sends data to this subscription.
-
-  void _sendData(T data) {
-    _onData(data);
-  }
-
-  void _sendError(error) {
-    _onError(error);
-    if (_cancelOnError) {
-      _subscription.cancel();
-      _subscription = null;
-    }
-  }
-
-  void _sendDone() {
-    // If the transformation sends a done signal, we stop the subscription.
-    if (_subscription != null) {
-      _subscription.cancel();
-      _subscription = null;
-    }
-    _onDone();
-  }
-
   // Methods used as listener on source subscription.
 
   // TODO(ahe): Restore type when feature is implemented in dart2js
   // checked mode. http://dartbug.com/7733
   void _handleData(/*S*/ data) {
     _stream._handleData(data, this);
   }
 
   void _handleError(error) {
     _stream._handleError(error, this);
@@ skipping 11 matching lines @@
 // -------------------------------------------------------------------
 
 typedef bool _Predicate<T>(T value);
 
 class _WhereStream<T> extends _ForwardingStream<T, T> {
   final _Predicate<T> _test;
 
   _WhereStream(Stream<T> source, bool test(T value))
       : _test = test, super(source);
 
-  void _handleData(T inputEvent, _EventOutputSink<T> sink) {
+  void _handleData(T inputEvent, _EventSink<T> sink) {
     bool satisfies;
     try {
       satisfies = _test(inputEvent);
     } catch (e, s) {
-      sink._sendError(_asyncError(e, s));
+      sink._addError(_asyncError(e, s));
       return;
     }
     if (satisfies) {
-      sink._sendData(inputEvent);
+      sink._add(inputEvent);
     }
   }
 }
 
 
 typedef T _Transformation<S, T>(S value);
 
 /**
  * A stream pipe that converts data events before passing them on.
  */
 class _MapStream<S, T> extends _ForwardingStream<S, T> {
   final _Transformation _transform;
 
   _MapStream(Stream<S> source, T transform(S event))
       : this._transform = transform, super(source);
 
-  void _handleData(S inputEvent, _EventOutputSink<T> sink) {
+  void _handleData(S inputEvent, _EventSink<T> sink) {
     T outputEvent;
     try {
       outputEvent = _transform(inputEvent);
     } catch (e, s) {
-      sink._sendError(_asyncError(e, s));
+      sink._addError(_asyncError(e, s));
       return;
     }
-    sink._sendData(outputEvent);
+    sink._add(outputEvent);
   }
 }
 
 /**
  * A stream pipe that converts data events before passing them on.
  */
 class _ExpandStream<S, T> extends _ForwardingStream<S, T> {
   final _Transformation<S, Iterable<T>> _expand;
 
   _ExpandStream(Stream<S> source, Iterable<T> expand(S event))
       : this._expand = expand, super(source);
 
-  void _handleData(S inputEvent, _EventOutputSink<T> sink) {
+  void _handleData(S inputEvent, _EventSink<T> sink) {
     try {
       for (T value in _expand(inputEvent)) {
-        sink._sendData(value);
+        sink._add(value);
       }
     } catch (e, s) {
       // If either _expand or iterating the generated iterator throws,
       // we abort the iteration.
-      sink._sendError(_asyncError(e, s));
+      sink._addError(_asyncError(e, s));
     }
   }
 }
 
 
 typedef void _ErrorTransformation(error);
 typedef bool _ErrorTest(error);
 
 /**
  * A stream pipe that converts or disposes error events
  * before passing them on.
  */
 class _HandleErrorStream<T> extends _ForwardingStream<T, T> {
   final _ErrorTransformation _transform;
   final _ErrorTest _test;
 
   _HandleErrorStream(Stream<T> source,
                     void transform(event),
                     bool test(error))
       : this._transform = transform, this._test = test, super(source);
 
-  void _handleError(Object error, _EventOutputSink<T> sink) {
+  void _handleError(Object error, _EventSink<T> sink) {
     bool matches = true;
     if (_test != null) {
       try {
         matches = _test(error);
       } catch (e, s) {
-        sink._sendError(_asyncError(e, s));
+        sink._addError(_asyncError(e, s));
         return;
       }
     }
     if (matches) {
       try {
         _transform(error);
       } catch (e, s) {
-        sink._sendError(_asyncError(e, s));
+        sink._addError(_asyncError(e, s));
         return;
       }
     } else {
-      sink._sendError(error);
+      sink._addError(error);
     }
   }
 }
 
 
 class _TakeStream<T> extends _ForwardingStream<T, T> {

[floitsch, 2013/05/24 15:53:17]

In theory it should be the subscription that does the counting. Since the stream
is single subscription it doesn't matter, though.

   int _remaining;
 
   _TakeStream(Stream<T> source, int count)
       : this._remaining = 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);
   }
 
-  void _handleData(T inputEvent, _EventOutputSink<T> sink) {
+  void _handleData(T inputEvent, _EventSink<T> sink) {
     if (_remaining > 0) {
-      sink._sendData(inputEvent);
+      sink._add(inputEvent);
       _remaining -= 1;
       if (_remaining == 0) {
         // Closing also unsubscribes all subscribers, which unsubscribes
         // this from source.
-        sink._sendDone();
+        sink._close();
       }
     }
   }
 }
 
 
 class _TakeWhileStream<T> extends _ForwardingStream<T, T> {
   final _Predicate<T> _test;
 
   _TakeWhileStream(Stream<T> source, bool test(T value))
       : this._test = test, super(source);
 
-  void _handleData(T inputEvent, _EventOutputSink<T> sink) {
+  void _handleData(T inputEvent, _EventSink<T> sink) {
     bool satisfies;
     try {
       satisfies = _test(inputEvent);
     } catch (e, s) {
-      sink._sendError(_asyncError(e, s));
+      sink._addError(_asyncError(e, s));
       // The test didn't say true. Didn't say false either, but we stop anyway.
-      sink._sendDone();
+      sink._close();
       return;
     }
     if (satisfies) {
-      sink._sendData(inputEvent);
+      sink._add(inputEvent);
     } else {
-      sink._sendDone();
+      sink._close();
     }
   }
 }
 
 class _SkipStream<T> extends _ForwardingStream<T, T> {
   int _remaining;
 
   _SkipStream(Stream<T> source, int count)
       : this._remaining = 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);
   }
 
-  void _handleData(T inputEvent, _EventOutputSink<T> sink) {
+  void _handleData(T inputEvent, _EventSink<T> sink) {
     if (_remaining > 0) {
       _remaining--;
       return;
     }
-    return sink._sendData(inputEvent);
+    return sink._add(inputEvent);
   }
 }
 
 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);
 
-  void _handleData(T inputEvent, _EventOutputSink<T> sink) {
+  void _handleData(T inputEvent, _EventSink<T> sink) {
     if (_hasFailed) {
-      sink._sendData(inputEvent);
+      sink._add(inputEvent);
       return;
     }
     bool satisfies;
     try {
       satisfies = _test(inputEvent);
     } catch (e, s) {
-      sink._sendError(_asyncError(e, s));
+      sink._addError(_asyncError(e, s));
       // A failure to return a boolean is considered "not matching".
       _hasFailed = true;
       return;
     }
     if (!satisfies) {
       _hasFailed = true;
-      sink._sendData(inputEvent);
+      sink._add(inputEvent);
     }
   }
 }
 
 typedef bool _Equality<T>(T a, T b);
 
 class _DistinctStream<T> extends _ForwardingStream<T, T> {
   static var _SENTINEL = new Object();
 
   _Equality<T> _equals;
   var _previous = _SENTINEL;
 
   _DistinctStream(Stream<T> source, bool equals(T a, T b))
       : _equals = equals, super(source);
 
-  void _handleData(T inputEvent, _EventOutputSink<T> sink) {
+  void _handleData(T inputEvent, _EventSink<T> sink) {
     if (identical(_previous, _SENTINEL)) {
       _previous = inputEvent;
-      return sink._sendData(inputEvent);
+      return sink._add(inputEvent);
     } else {
       bool isEqual;
       try {
         if (_equals == null) {
           isEqual = (_previous == inputEvent);
         } else {
           isEqual = _equals(_previous, inputEvent);
         }
       } catch (e, s) {
-        sink._sendError(_asyncError(e, s));
+        sink._addError(_asyncError(e, s));
         return null;
       }
       if (!isEqual) {
-        sink._sendData(inputEvent);
+        sink._add(inputEvent);
         _previous = inputEvent;
       }
     }
   }
 }
 
 // Stream transformations and event transformations.
 
 typedef void _TransformDataHandler<S, T>(S data, EventSink<T> sink);
 typedef void _TransformErrorHandler<T>(data, EventSink<T> sink);
@@ skipping 48 matching lines @@
 
   void handleError(error, EventSink<T> sink) {
     _handleError(error, sink);
   }
 
   void handleDone(EventSink<T> sink) {
     _handleDone(sink);
   }
 }