Add ClosedStreamSink, and *Completer.setError, and StreamSinkCompleter.fromFuture.

lib/src/stream_completer.dart

 // Copyright (c) 2015, 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.
 
 library async.stream_completer;
 
 import "dart:async";
 
 /// A single-subscription [stream] where the contents are provided later.
 ///
@@ skipping 21 matching lines @@
   ///
   /// This creates a stream using a stream completer,
   /// and sets the source stream to the result of the future when the
   /// future completes.
   ///
   /// If the future completes with an error, the returned stream will
   /// instead contain just that error.
   static Stream fromFuture(Future<Stream> streamFuture) {
     var completer = new StreamCompleter();
     streamFuture.then(completer.setSourceStream,
-                      onError: (e, s) {
-                        completer.setSourceStream(streamFuture.asStream());
-                      });
+        onError: completer.setError);
     return completer.stream;
   }
 
   /// The stream of this completer.
   ///
   /// This stream is always a single-subscription stream.
   ///
   /// When a source stream is provided, its events will be forwarded to
   /// listeners on this stream.
   ///
@@ skipping 14 matching lines @@
   /// normal subscription, and can be paused or canceled, but it won't
   /// produce any events until a source stream is provided.
   ///
   /// If the `stream` subscription is canceled before a source stream is set,
   /// the source stream will be listened to and immediately canceled again.
   ///
   /// Otherwise, when the source stream is then set,
   /// it is immediately listened to, and its events are forwarded to the
   /// existing subscription.
   ///
-  /// Either [setSourceStream] or [setEmpty] may be called at most once.
-  /// Trying to call either of them again will fail.
+  /// Any one of [setSourceStream], [setEmpty], and [setError] may be called at
+  /// most once. Trying to call any of them again will fail.
   void setSourceStream(Stream<T> sourceStream) {
     if (_stream._isSourceStreamSet) {
       throw new StateError("Source stream already set");
     }
     _stream._setSourceStream(sourceStream);
   }
 
   /// Equivalent to setting an empty stream using [setSourceStream].
   ///
-  /// Either [setSourceStream] or [setEmpty] may be called at most once.
-  /// Trying to call either of them again will fail.
+  /// Any one of [setSourceStream], [setEmpty], and [setError] may be called at
+  /// most once. Trying to call any of them again will fail.
   void setEmpty() {
     if (_stream._isSourceStreamSet) {
       throw new StateError("Source stream already set");
     }
     _stream._setEmpty();
   }
+
+  /// Completes this to a stream that emits [error] and then closes.
+  ///
+  /// This is useful when the process of creating the data for the stream fails.
+  ///
+  /// Any one of [setSourceStream], [setEmpty], and [setError] may be called at
+  /// most once. Trying to call any of them again will fail.
+  void setError(error, [StackTrace stackTrace]) {
+    setSourceStream(new Stream.fromFuture(new Future.error(error, stackTrace)));
+  }
 }
 
 /// Stream completed by [StreamCompleter].
 class _CompleterStream<T> extends Stream<T> {
   /// Controller for an intermediate stream.
   ///
   /// Created if the user listens on this stream before the source stream
   /// is set, or if using [_setEmpty] so there is no source stream.
   StreamController _controller;
 
@@ skipping 63 matching lines @@
     _sourceStream = _controller.stream;  // Mark stream as set.
     _controller.close();
   }
 
   // Creates the [_controller].
   void _createController() {
     assert(_controller == null);
     _controller = new StreamController<T>(sync: true);
   }
 }