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 607 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> {
+
+  void _onSubscriptionStateChange() {
+    if (_hasSubscribers && !_isClosed) {
+      new Timer(0, (_) {
+        _handlePendingEvents();
+      });
+    }
+  }
+
+  void _generateNextEvent();
+
+  void _handlePendingEvents() {
+    while (!_isPaused) {
+      super._handlePendingEvents();
+      if (!_isPaused && !_isClosed) {
+        _generateNextEvent();
+      }
+    }
+  }
+}
+
+
+/** 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 370 matching lines @@
     if (_isComplete) {
       throw new StateError("Subscription has been canceled.");
     }
     if (_timer != null) {
       _timer.cancel();
       _timer = null;
     }
     _pauseCount = 0;
   }
 }