Make StreamSubscription be the active part of a stream.

sdk/lib/async/stream.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.async;
 
 // -------------------------------------------------------------------
 // Core Stream types
 // -------------------------------------------------------------------
 
@@ skipping 44 matching lines @@
 abstract class Stream<T> {
   Stream();
 
   /**
    * Creates a new single-subscription stream from the future.
    *
    * When the future completes, the stream will fire one event, either
    * data or error, and then close with a done-event.
    */
   factory Stream.fromFuture(Future<T> future) {
-    _StreamImpl<T> stream = new _SingleStreamImpl<T>();
+    StreamController<T> controller = new StreamController<T>();
     future.then((value) {
-        stream._add(value);
-        stream._close();
+        controller.add(value);
+        controller.close();
       },
       onError: (error) {
-        stream._addError(error);
-        stream._close();
+        controller.addError(error);
+        controller.close();
       });
-    return stream;
+    return controller.stream;
   }
 
   /**
    * Creates a single-subscription stream that gets its data from [data].
    */
   factory Stream.fromIterable(Iterable<T> data) {
-    _PendingEvents iterableEvents = new _IterablePendingEvents<T>(data);
-    return new _GeneratedSingleStreamImpl<T>(iterableEvents);
+    return new _GeneratedStreamImpl<T>(
+        () => new _IterablePendingEvents<T>(data));
   }
 
   /**
    * Creates a stream that repeatedly emits events at [period] intervals.
    *
    * The event values are computed by invoking [computation]. The argument to
    * this callback is an integer that starts with 0 and is incremented for
    * every event.
    *
    * If [computation] is omitted the event values will all be `null`.
@@ skipping 935 matching lines @@
  *     someTypeStream.transform(new DoubleTransformer<Type>());
  *
  * The default implementations of the "handle" methods forward
  * the events unmodified. If using the default [handleData] the generic type [T]
  * needs to be assignable to [S].
  */
 abstract class StreamEventTransformer<S, T> implements StreamTransformer<S, T> {
   const StreamEventTransformer();
 
   Stream<T> bind(Stream<S> source) {
-    // Hackish way of buffering data that goes out of the event-transformer.
-    // TODO(floitsch): replace this with a correct solution.
-    Stream transformingStream = new EventTransformStream<S, T>(source, this);
-    StreamController controller;
-    StreamSubscription subscription;
-    controller = new StreamController<T>(
-        onListen: () {
-          subscription = transformingStream.listen(
-              controller.add,
-              onError: controller.addError,
-              onDone: controller.close);
-        },
-        onPause: () => subscription.pause(),
-        onResume: () => subscription.resume(),
-        onCancel: () => subscription.cancel());
-    return controller.stream;
+    return new EventTransformStream<S, T>(source, this);
   }
 
   /**
    * Act on incoming data event.
    *
    * The method may generate any number of events on the sink, but should
    * not throw.
    */
   void handleData(S event, EventSink<T> sink) {
     var data = event;
@@ skipping 34 matching lines @@
   final Stream<S> _source;
   final StreamEventTransformer _transformer;
   EventTransformStream(Stream<S> source,
                        StreamEventTransformer<S, T> transformer)
       : _source = source, _transformer = transformer;
 
   StreamSubscription<T> listen(void onData(T data),
                                { void onError(error),
                                  void onDone(),
                                  bool cancelOnError }) {
+    if (onData == null) onData = _nullDataHandler;
+    if (onError == null) onError = _nullErrorHandler;
+    if (onDone == null) onDone = _nullDoneHandler;
     cancelOnError = identical(true, cancelOnError);
     return new _EventTransformStreamSubscription(_source, _transformer,
                                                  onData, onError, onDone,
                                                  cancelOnError);
   }
 }
 
 class _EventTransformStreamSubscription<S, T>
-    extends _BaseStreamSubscription<T>
-    implements _EventOutputSink<T> {
+    extends _BufferingStreamSubscription<T> {
   /** The transformer used to transform events. */
   final StreamEventTransformer<S, T> _transformer;
-  /** Whether to unsubscribe when emitting an error. */
-  final bool _cancelOnError;
+
   /** Whether this stream has sent a done event. */
   bool _isClosed = false;
+
   /** Source of incoming events. */
   StreamSubscription<S> _subscription;
+
   /** Cached EventSink wrapper for this class. */
   EventSink<T> _sink;
 
   _EventTransformStreamSubscription(Stream<S> source,
                                     this._transformer,
                                     void onData(T data),
                                     void onError(error),
                                     void onDone(),
-                                    this._cancelOnError)
-      : super(onData, onError, onDone) {
-    _sink = new _EventOutputSinkWrapper<T>(this);
+                                    bool cancelOnError)
+      : super(onData, onError, onDone, cancelOnError) {
+    _sink = new _EventSinkAdapter<T>(this);
     _subscription = source.listen(_handleData,
                                   onError: _handleError,
                                   onDone: _handleDone);
   }
 
   /** Whether this subscription is still subscribed to its source. */
   bool get _isSubscribed => _subscription != null;
 
-  void pause([Future pauseSignal]) {
-    if (_isSubscribed) _subscription.pause(pauseSignal);
+  void _onPause() {
+    if (_isSubscribed) _subscription.pause();
   }
 
-  void resume() {
+  void _onResume() {
     if (_isSubscribed) _subscription.resume();
   }
 
-  bool get isPaused => _isSubscribed ? _subscription.isPaused : false;
-
-  void cancel() {
+  void _onCancel() {
     if (_isSubscribed) {
       StreamSubscription subscription = _subscription;
       _subscription = null;
       subscription.cancel();
     }
     _isClosed = true;
   }
 
   void _handleData(S data) {
     try {
       _transformer.handleData(data, _sink);
     } catch (e, s) {
-      _sendError(_asyncError(e, s));
+      _addError(_asyncError(e, s));
     }
   }
 
   void _handleError(error) {
     try {
       _transformer.handleError(error, _sink);
     } catch (e, s) {
-      _sendError(_asyncError(e, s));
+      _addError(_asyncError(e, s));
     }
   }
 
   void _handleDone() {
     try {
       _subscription = null;
       _transformer.handleDone(_sink);
     } catch (e, s) {
-      _sendError(_asyncError(e, s));
+      _addError(_asyncError(e, s));
     }
   }
-
-  // EventOutputSink interface.
-  void _sendData(T data) {
-    if (_isClosed) return;
-    _onData(data);
-  }
-
-  void _sendError(error) {
-    if (_isClosed) return;
-    _onError(error);
-    if (_cancelOnError) {
-      cancel();
-    }
-  }
-
-  void _sendDone() {
-    if (_isClosed) throw new StateError("Already closed.");
-    _isClosed = true;
-    if (_isSubscribed) {
-      _subscription.cancel();
-      _subscription = null;
-    }
-    _onDone();
-  }
 }
 
-class _EventOutputSinkWrapper<T> extends EventSink<T> {
-  _EventOutputSink _sink;
-  _EventOutputSinkWrapper(this._sink);
+class _EventSinkAdapter<T> implements EventSink<T> {
+  _EventSink _sink;
+  _EventSinkAdapter(this._sink);
 
-  void add(T data) { _sink._sendData(data); }
-  void addError(error) { _sink._sendError(error); }
-  void close() { _sink._sendDone(); }
+  void add(T data) { _sink._add(data); }
+  void addError(error) { _sink._addError(error); }
+  void close() { _sink._close(); }
 }
+
+
+/**
+ * An [Iterable] like interface for the values of a [Stream].
+ *
+ * This wraps a [Stream] and a subscription on the stream. It listens
+ * on the stream, and completes the future returned by [moveNext] when the
+ * next value becomes available.
+ */
+abstract class StreamIterator<T> {
+
+  /** Create a [StreamIterator] on [stream]. */
+  factory StreamIterator(Stream<T> stream)
+      // TODO(lrn): use redirecting factory constructor when type
+      // arguments are supported.
+      => new _StreamIteratorImpl<T>(stream);
+
+  /**
+   * Wait for the next stream value to be available.
+   *
+   * It is not allowed to call this function again until the future has
+   * completed. If the returned future completes with anything except `true`,
+   * the iterator is done, and no new value will ever be available.
+   *
+   * The future may complete with an error, if the stream produces an error.
+   */
+  Future<bool> moveNext();
+
+  /**
+   * The current value of the stream.
+   *
+   * Only valid when the future returned by [moveNext] completes with `true`

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

"valid" or non-null?

+   * as value, and only until the next call to [moveNext].
+   */
+  T get current;
+
+  /**
+   * Cancels the stream iterator (and the underlying stream subscription) early.
+   *
+   * The stream iterator is automatically cancelled if the [moveNext] future
+   * completes with either `false` or an error.
+   *
+   * If a [moveNext] call has been made, it will complete with `false` as value,
+   * as will all further calls to [moveNext].
+   *
+   * If you need to stop listening for values before the stream iterator is
+   * automatically closed, you must call [cancel] to ensure that the stream
+   * is properly closed.
+   */
+  void cancel();
+}