Make Stream.fromIterable only accept a single listen call.

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;
 
 /** Abstract and private interface for a place to put events. */
 abstract class _EventSink<T> {
   void _add(T data);
   void _addError(Object error, StackTrace stackTrace);
@@ skipping 470 matching lines @@
 
   /** Hook called when the subscription has been created. */
   void _onListen(StreamSubscription subscription) {}
 }
 
 typedef _PendingEvents _EventGenerator();
 
 /** Stream that generates its own events. */
 class _GeneratedStreamImpl<T> extends _StreamImpl<T> {
   final _EventGenerator _pending;
+  bool _isUsed = false;
   /**
    * Initializes the stream to have only the events provided by a
    * [_PendingEvents].
    *
    * A new [_PendingEvents] must be generated for each listen.
    */
   _GeneratedStreamImpl(this._pending);
 
   StreamSubscription _createSubscription(bool cancelOnError) {
+    if (_isUsed) throw new StateError("Stream has already been listened to.");
+    _isUsed = true;
     _BufferingStreamSubscription<T> subscription =
          new _BufferingStreamSubscription(cancelOnError);
     subscription._setPendingEvents(_pending());
     return subscription;
   }
 }
 
 
 /** Pending events object that gets its events from an [Iterable]. */
 class _IterablePendingEvents<T> extends _PendingEvents {
@@ skipping 515 matching lines @@
       _Future<bool> hasNext = _futureOrPrefetch;
       _clear();
       hasNext._complete(false);
       return;
     }
     _subscription.pause();
     _futureOrPrefetch = null;
     _state = _STATE_EXTRA_DONE;
   }
 }