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

lib/src/stream_sink_completer.dart

 // Copyright (c) 2016, 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_sink_completer;
 
 import 'dart:async';
 
+import 'null_stream_sink.dart';
+
 /// A [sink] where the destination is provided later.
 ///
 /// The [sink] is a normal sink that you can add events to to immediately, but
 /// until [setDestinationSink] is called, the events will be buffered.
 ///
 /// The same effect can be achieved by using a [StreamController] and adding it
 /// to the sink using [Sink.addStream] when the destination sink is ready. This
 /// class attempts to shortcut some of the overhead when possible. For example,
 /// if the [sink] only has events added after the destination sink has been set,
 /// those events are added directly to the sink.
 class StreamSinkCompleter<T> {
   /// The sink for this completer.
   ///
   /// When a destination sink is provided, events that have been passed to the
   /// sink will be forwarded to the destination.
   ///
   /// Events can be added to the sink either before or after a destination sink
   /// is set.
   final StreamSink<T> sink = new _CompleterSink<T>();
 
   /// Returns [sink] typed as a [_CompleterSink].
   _CompleterSink<T> get _sink => sink;
 
+  /// Convert a `Future<StreamSink>` to a `StreamSink`.
+  ///
+  /// This creates a sink using a sink completer, and sets the destination sink
+  /// to the result of the future when the future completes.
+  ///
+  /// If the future completes with an error, the returned sink will instead
+  /// be closed. Its [Sink.done] future will contain the error.
+  static StreamSink fromFuture(Future<StreamSink> sinkFuture) {
+    var completer = new StreamSinkCompleter();
+    sinkFuture.then(completer.setDestinationSink,
+        onError: completer.setError);
+    return completer.sink;
+  }
+
   /// Sets a sink as the destination for events from the [StreamSinkCompleter]'s
   /// [sink].
   ///
   /// The completer's [sink] will act exactly as [destinationSink].
   ///
   /// If the destination sink is set before events are added to [sink], further
   /// events are forwarded directly to [destinationSink].
   ///
   /// If events are added to [sink] before setting the destination sink, they're
   /// buffered until the destination is available.
   ///
   /// A destination sink may be set at most once.
+  ///
+  /// Either of [setDestinationSink] or [setError] may be called at most once.
+  /// Trying to call either of them again will fail.
   void setDestinationSink(StreamSink<T> destinationSink) {
     if (_sink._destinationSink != null) {
       throw new StateError("Destination sink already set");
     }
     _sink._setDestinationSink(destinationSink);
   }
+
+  /// Completes this to a closed sink whose [Sink.done] future emits [error].
+  ///
+  /// This is useful when the process of loading the sink fails.
+  ///
+  /// Either of [setDestinationSink] or [setError] may be called at most once.
+  /// Trying to call either of them again will fail.
+  void setError(error, [StackTrace stackTrace]) {
+    setDestinationSink(new NullStreamSink.error(error, stackTrace));
+  }
 }
 
 /// [StreamSink] completed by [StreamSinkCompleter].
 class _CompleterSink<T> implements StreamSink<T> {
   /// Controller for an intermediate sink.
   ///
   /// Created if the user adds events to this sink before the destination sink
   /// is set.
   StreamController<T> _controller;
 
@@ skipping 82 matching lines @@
           .catchError((_) {});
     }
 
     // If the user has already asked when the sink is done, connect the sink's
     // done callback to that completer.
     if (_doneCompleter != null) {
       _doneCompleter.complete(sink.done);
     }
   }
 }