Add Stream.fromIterable

sdk/lib/async/stream_impl.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;
 
 // States shared by single/multi stream implementations.
 
 /// Initial and default state where the stream can receive and send events.
 const int _STREAM_OPEN = 0;
@@ skipping 424 matching lines @@
         this, onData, onError, onDone, unsubscribeOnError);
   }
 
   void _addListener(_StreamSubscriptionImpl subscription) {
     if (_hasSubscribers) {
       throw new StateError("Stream has already subscriber.");
     }
     _subscriber = subscription;
     subscription._setSubscribed(0);
     _onSubscriptionStateChange();
-    // TODO(floitsch): Should this be delayed?
-    _handlePendingEvents();
+    if (_hasPendingEvent) {
+      new Timer(0, (_) {
+        _handlePendingEvents();
+      });
+    }
   }
 
   /**
    * Handle a cancel requested from a [_StreamSubscriptionImpl].
    *
    * This method is called from [_StreamSubscriptionImpl.cancel].
    *
    * If an event is currently firing, the cancel is delayed
    * until after the subscriber has received the event.
    */
@@ skipping 170 matching lines @@
    * This is a low-level action that doesn't call [_onSubscriptionStateChange].
    * or [_onPauseStateChange].
    */
   void _removeListener(_StreamListener listener) {
     int pauseCount = listener._setUnsubscribed();
     _updatePauseCount(-pauseCount);
     _InternalLinkList.remove(listener);
   }
 }
 
+
+/** Abstract superclass for streams that generate their own events. */
+abstract class _GeneratedSingleStreamImpl<T> extends _SingleStreamImpl<T> {
+  bool _isHandlingPendingEvents = false;
+  bool get _hasPendingEvent => !_isClosed;
+
+  /**
+   * Generate one (or possibly more) new events.
+   *
+   * The events should be added to the stream using [_add], [_signalError] and
+   * [_close].
+   */
+  void _generateNextEvent();
+
+  void _handlePendingEvents() {
+    // Avoid reentry from _add/_signalError/_close potentially called
+    // from _generateNextEvent.
+    if (_isHandlingPendingEvents) return;
+    _isHandlingPendingEvents = true;
+    while (!_isPaused && !_isClosed) {
+      // Call super's handle event in case _generateNextEvent generates
+      // more than one event, and the following ones are delayed.
+      super._handlePendingEvents();
+      if (!_isPaused && !_isClosed) {
+        _generateNextEvent();
+      }
+    }
+    _isHandlingPendingEvents = false;
+  }
+}
+
+
+/** Stream that gets its events from an [Iterable]. */
+class _IterableSingleStreamImpl<T> extends _GeneratedSingleStreamImpl<T> {
+  Iterator<T> _iterator;
+
+  _IterableSingleStreamImpl(Iterable<T> data) : _iterator = data.iterator;
+
+  void _generateNextEvent() {
+    try {
+      if (_iterator.moveNext()) {
+        _add(_iterator.current);
+        return;
+      }
+    } catch (e, s) {
+      _signalError(new AsyncError(e, s));
+    }
+    _close();
+  }
+}
+
+
 /**
  * The subscription class that the [StreamController] uses.
  *
  * The [StreamController.createSubscription] method should
  * create an object of this type, or another subclass of [_StreamListener].
  * A subclass of [StreamController] can specify which subclass
  * of [_StreamSubscriptionImpl] it uses by overriding
  * [StreamController.createSubscription].
  *
  * The subscription is in one of three states:
@@ skipping 372 matching lines @@
     if (_isComplete) {
       throw new StateError("Subscription has been canceled.");
     }
     if (_timer != null) {
       _timer.cancel();
       _timer = null;
     }
     _pauseCount = 0;
   }
 }