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

lib/src/stream_completer.dart

Index: lib/src/stream_completer.dart
diff --git a/lib/src/stream_completer.dart b/lib/src/stream_completer.dart
index c343e6e7bd2330ccda59e9c793f96b0de3fbed50..0945b6a8733cae7d185cf5a436524e8c96b78dfa 100644
--- a/lib/src/stream_completer.dart
+++ b/lib/src/stream_completer.dart
@@ -39,9 +39,7 @@ class StreamCompleter<T> {
   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;
   }
 
@@ -76,8 +74,8 @@ class StreamCompleter<T> {
   /// 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 of [setSourceStream], [setEmpty], and [setError] may be called at most

[Lasse Reichstein Nielsen, 2016/01/22 09:13:47]

Any one of ...

(The current writing is unclear on whether you can call setEmpty and setError
once each).

[nweiz, 2016/01/25 22:39:25]

Done.

+  /// 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");
@@ -87,14 +85,24 @@ class StreamCompleter<T> {
 
   /// 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 of [setSourceStream], [setEmpty], and [setError] may be called at most

[Lasse Reichstein Nielsen, 2016/01/22 09:13:46]

Any one of ...

[nweiz, 2016/01/25 22:39:25]

Done.

+  /// 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 loading the stream fails.

[Lasse Reichstein Nielsen, 2016/01/22 09:13:46]

loading the stream -> creating the data for the stream

(loading is too specific, and not something you do to a stream).

[nweiz, 2016/01/25 22:39:25]

Done.

+  ///
+  /// Any of [setSourceStream], [setEmpty], and [setError] may be called at most

[Lasse Reichstein Nielsen, 2016/01/22 09:13:46]

Any one ..

[nweiz, 2016/01/25 22:39:25]

Done.

+  /// once. Trying to call any of them again will fail.
+  void setError(error, [StackTrace stackTrace]) {

[Lasse Reichstein Nielsen, 2016/01/22 09:13:47]

Good idea!

[nweiz, 2016/01/25 22:39:25]

:D

+    setSourceStream(new Stream.fromFuture(new Future.error(error, stackTrace)));
+  }
 }
 
 /// Stream completed by [StreamCompleter].