Revert until Windows crash is debugged.

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) {
-    StreamController<T> controller = new StreamController<T>();
+    _StreamImpl<T> stream = new _SingleStreamImpl<T>();
     future.then((value) {
-        controller.add(value);
-        controller.close();
+        stream._add(value);
+        stream._close();
       },
       onError: (error) {
-        controller.addError(error);
-        controller.close();
+        stream._addError(error);
+        stream._close();
       });
-    return controller.stream;
+    return stream;
   }
 
   /**
    * Creates a single-subscription stream that gets its data from [data].
    */
   factory Stream.fromIterable(Iterable<T> data) {
-    return new _GeneratedStreamImpl<T>(
-        () => new _IterablePendingEvents<T>(data));
+    _PendingEvents iterableEvents = new _IterablePendingEvents<T>(data);
+    return new _GeneratedSingleStreamImpl<T>(iterableEvents);
   }
 
   /**
    * 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 52 matching lines @@
    * Reports whether this stream is a broadcast stream.
    */
   bool get isBroadcast => false;
 
   /**
    * Returns a multi-subscription stream that produces the same events as this.
    *
    * If this stream is single-subscription, return a new stream that allows
    * multiple subscribers. It will subscribe to this stream when its first
    * subscriber is added, and unsubscribe again when the last subscription is
-   * canceled.
+   * cancelled.
    *
    * If this stream is already a broadcast stream, it is returned unmodified.
    */
   Stream<T> asBroadcastStream() {
     if (isBroadcast) return this;
     return new _SingleStreamMultiplexer<T>(this);
   }
 
   /**
    * Adds a subscription to this stream.
@@ skipping 876 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) {
-    return new EventTransformStream<S, T>(source, this);
+    // 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;
   }
 
   /**
    * 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 _BufferingStreamSubscription<T> {
+    extends _BaseStreamSubscription<T>
+    implements _EventOutputSink<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(),
-                                    bool cancelOnError)
-      : super(onData, onError, onDone, cancelOnError) {
-    _sink = new _EventSinkAdapter<T>(this);
+                                    this._cancelOnError)
+      : super(onData, onError, onDone) {
+    _sink = new _EventOutputSinkWrapper<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 _onPause() {
-    if (_isSubscribed) _subscription.pause();
+  void pause([Future pauseSignal]) {
+    if (_isSubscribed) _subscription.pause(pauseSignal);
   }
 
-  void _onResume() {
+  void resume() {
     if (_isSubscribed) _subscription.resume();
   }
 
-  void _onCancel() {
+  bool get isPaused => _isSubscribed ? _subscription.isPaused : false;
+
+  void cancel() {
     if (_isSubscribed) {
       StreamSubscription subscription = _subscription;
       _subscription = null;
       subscription.cancel();
     }
     _isClosed = true;
   }
 
   void _handleData(S data) {
     try {
       _transformer.handleData(data, _sink);
     } catch (e, s) {
-      _addError(_asyncError(e, s));
+      _sendError(_asyncError(e, s));
     }
   }
 
   void _handleError(error) {
     try {
       _transformer.handleError(error, _sink);
     } catch (e, s) {
-      _addError(_asyncError(e, s));
+      _sendError(_asyncError(e, s));
     }
   }
 
   void _handleDone() {
     try {
       _subscription = null;
       _transformer.handleDone(_sink);
     } catch (e, s) {
-      _addError(_asyncError(e, s));
+      _sendError(_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 _EventSinkAdapter<T> implements EventSink<T> {
-  _EventSink _sink;
-  _EventSinkAdapter(this._sink);
+class _EventOutputSinkWrapper<T> extends EventSink<T> {
+  _EventOutputSink _sink;
+  _EventOutputSinkWrapper(this._sink);
 
-  void add(T data) { _sink._add(data); }
-  void addError(error) { _sink._addError(error); }
-  void close() { _sink._close(); }
+  void add(T data) { _sink._sendData(data); }
+  void addError(error) { _sink._sendError(error); }
+  void close() { _sink._sendDone(); }
 }
-
-
-/**
- * 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.
- *
- * NOTICE: This is a tentative design. This class may change.
- */
-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`
-   * 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 canceled 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();
-}