Add a StreamSinkCompleter class.

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';
+
+/// 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> {

[Lasse Reichstein Nielsen, 2016/01/21 07:28:47]

I don't think I like "completer" for this particular use.
How about "StreamSinkBuffer" or "DelayedStreamSink"?
(Or something better!)

I think we have used Completer in other cases, but in hindsight I'm not sure
that was a good idea - it's a fairly nondescriptive name and Future got the
primary use of it. I don't think this class is close enough to what the future
completer does. I think the difference is that Completer and StreamCompleter
will "complete" with the incoming result, while this class lets you set the
output - the result isn't "completed" afterwards, it's just getting started.

[nweiz, 2016/01/21 20:59:05]

I really think consistency with parallel APIs is important here. I intend to use
this alongside a similar use of StreamCompleter, and I suspect many uses will be
similar. Having the names be similar helps emphasize their parallel purpose.

In general, I also think that broadening an existing term to cover variations of
the concept is often better than creating a new term. It's easier for users to
understand "Completer" to mean "supplying a concrete object after the fact" than
it is for them to learn a use of "Buffered" or "Delayed" to describe a class
whose usage isn't closely related to any other use of those terms in Dart's
libraries.

[Lasse Reichstein Nielsen, 2016/01/22 12:49:27]

Acknowledged. 
It doesn't fit my mental model of what "completing" means (putting in the
*result*), but I guess it's consistent in its own way.

+  /// 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.

[Lasse Reichstein Nielsen, 2016/01/21 07:28:47]

Do you want to promise how the events are forwarded (as timer events, as
microtasks, or all in one go?).
Probably not - it's safer to not say anything, but it will mean that you can't
change the timing later because people will depend on the actual behavior if you
don't explicitly say that the behavior is not promised.

(I *really* want to make stream-controllers deliver their buffered events using
Timer(0) instead of scheduleMicrotask, but that's probably a breaking change at
this point).

[nweiz, 2016/01/21 20:59:05]

I don't think we can provide any stronger guarantees than [StreamController]
does already. Since it's unspecified, I think we have to leave this unspecified
as well.
Why, out of curiosity? That seems inconsistent with Futures.

[Lasse Reichstein Nielsen, 2016/01/22 12:49:27]

That's actually bad - leaving something unspecified implicitly means that users
can only depend on what it actually does. If we say explicitly that there are no
promises, then users won't be able to (reasonably) depend on the actual
behavior. 
The thing is that if a stream has a lot of buffered events, the code handling
them will become completely interleaved. If instead it waited until the
microtask queue is empty to release the next event, then the event handling code
will be nicely separated into chunks related to each event.

The total amount of work would be the same, I just like the ordering better when
the microtasks triggered by an event gets to complete before the next event is
fired.

+  ///
+  /// Events can be added to the sink either before or after a destination sink
+  /// is set.
+  StreamSink<T> get sink => _sink;
+  final _sink = new _CompleterSink<T>();

[Lasse Reichstein Nielsen, 2016/01/21 07:28:47]

Just make it one field:
  final StreamSink<T> sink = new _CompleterSink<T>();

You may need to assign it to a local variable for typing when using:
  _CompleterSink sink = _sink;
  ...
since _sink is dynamic.

[nweiz, 2016/01/21 20:59:05]

Done.

+
+  /// Set a sink as the destination for events from the [StreamSinkCompleter]'s

[Lasse Reichstein Nielsen, 2016/01/21 07:28:47]

Set -> Sets

[nweiz, 2016/01/21 20:59:05]

Done.

+  /// [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.
+  void setDestinationSink(StreamSink<T> destinationSink) {

[Lasse Reichstein Nielsen, 2016/01/21 07:28:47]

Make this a setter.

[nweiz, 2016/01/21 20:59:05]

That's inconsistent with StreamCompleter (and Completer for that matter). Also,
the design guide says not to use a setter without a corresponding getter
(https://www.dartlang.org/effective-dart/design/#dont-define-a-setter-without-...),
and I don't think a getter makes sense here.

[Lasse Reichstein Nielsen, 2016/01/22 12:49:27]

Argument accepted.

It is probably better as a method than a setter. Especially when it can only set
once.

+    if (_sink._destinationSink != null) {
+      throw new StateError("Destination sink already set");
+    }
+    _sink._setDestinationSink(destinationSink);

[Lasse Reichstein Nielsen, 2016/01/21 07:28:47]

Can you find another name than this (in particular, one that doesn't start with
"set")?
I'm not coming up with any, but this name looks like something which should be a
setter.
If you stick with "Completer" in the class name, maybe name it "completeSink"
(which sounds bad, so maybe it isn't a completer :)

[nweiz, 2016/01/21 20:59:05]

This is consistent with StreamCompleter (which you wrote ;) ). I think it's okay
to have methods start with "set" when the setting is meant to have side effects
rather than making a property accessible.

That said, if you feel strongly that "set" is wrong, I can send out a follow-up
CL that changes it to "complete" (which I don't mind much) and deprecates
StreamCompleter.setStream() in favor of StreamCompleter.completeStream().

[Lasse Reichstein Nielsen, 2016/01/22 12:49:27]

The problem is that "set" isn't wrong. It just looks like it should have been a
setter. But I think you are right that this is better than the alternatives.

+  }
+}
+
+/// [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;
+
+  /// Completer for [done].
+  ///
+  /// Created if the user requests the [done] future before the destination sink
+  /// is set.
+  Completer _doneCompleter;
+
+  /// Destination sink for the events added to this sink.
+  ///
+  /// Set when [StreamSinkCompleter.setDestinationSink] is called.
+  StreamSink<T> _destinationSink;
+
+  /// Whether events should be send directly to [_destinationSink], as opposed
+  /// to going through [_controller].
+  bool get _sendToSink => _controller == null && _destinationSink != null;

[Lasse Reichstein Nielsen, 2016/01/21 07:28:47]

Name sounds imperative. Maybe "_hasDirectSink" or something less awkward?

If you can't find a way to prefix an "is", "has", "will" or similar, maybe just
change to "_sendsToSink".

[nweiz, 2016/01/21 20:59:05]

Done.

+
+  Future get done {
+    if (_doneCompleter != null) return _doneCompleter.future;
+    if (_destinationSink == null) {
+      _doneCompleter = new Completer.sync();
+      return _doneCompleter.future;
+    }
+    return _destinationSink.done;
+  }
+
+  void add(T event) {
+    if (_sendToSink) {
+      _destinationSink.add(event);
+    } else {
+      _ensureController();
+      _controller.add(event);
+    }
+  }
+
+  void addError(error, [StackTrace stackTrace]) {
+    if (_sendToSink) {
+      _destinationSink.addError(error, stackTrace);
+    } else {
+      _ensureController();
+      _controller.addError(error, stackTrace);
+    }
+  }
+
+  Future addStream(Stream<T> stream) {
+    if (_sendToSink) return _destinationSink.addStream(stream);
+
+    _ensureController();
+    return _controller.addStream(stream, cancelOnError: false);
+  }
+
+  Future close() {
+    if (_sendToSink) {
+      _destinationSink.close();
+    } else {
+      _ensureController();
+      _controller.close();
+    }
+    return done;
+  }
+
+  /// Create [_controller] if it doesn't yet exist.
+  void _ensureController() {
+    if (_controller == null) _controller = new StreamController<T>(sync: true);

[Lasse Reichstein Nielsen, 2016/01/21 07:28:47]

Persionally, I'd drop the type parameter on the completely internal controller.
The user will see neither the controller nor the stream, so it doesn't matter,
and we have type parameters on the add method and setDestinationSink.

Completely personal preference.

[nweiz, 2016/01/21 20:59:05]

Done.

+  }
+
+  /// Sets the destination sink to which events from this sink will be provided.
+  ///
+  /// If set before the user adds events, events will be added directly to the
+  /// destination sink. If the user adds events earlier, an intermediate sink is
+  /// created using a stream controller, and the destination sink is linked to
+  /// it later.
+  void _setDestinationSink(StreamSink<T> sink) {
+    assert(_destinationSink == null);
+    _destinationSink = sink;
+
+    // If the user has already added data, it's buffered in the controller, so
+    // we add it to the sink.
+    if (_controller != null) {
+      // Catch any error that may come from [addStream] or [sink.close]. They'll
+      // be reported through [done] anyway.
+      sink
+          .addStream(_controller.stream)
+          .whenComplete(sink.close)
+          .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);
+    }
+  }
+}