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

lib/src/closed_stream_sink.dart

Index: lib/src/closed_stream_sink.dart
diff --git a/lib/src/closed_stream_sink.dart b/lib/src/closed_stream_sink.dart
new file mode 100644
index 0000000000000000000000000000000000000000..151f419a03828d09d54af40b8349ce45aa81b95c
--- /dev/null
+++ b/lib/src/closed_stream_sink.dart
@@ -0,0 +1,59 @@
+// 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.closed_stream_sink;
+
+import 'dart:async';
+
+/// A [StreamSink] that's created in a closed state.
+///
+/// The sink silently drops events until [close] is called, at which point it
+/// throws [StateError]s when events are added.

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

It confuses me that it's "in a closed state" before "close" is called.

It's basically a StreamSink that *ignores* data and error events.
Maybe name it:
  VoidStreamSink
  NullStreamSink (but that's using "null" as in /dev/null, not Null the type)
  DiscardingStreamSink
  IgnoreEventStreamSink
?

Or make it actually act as if close had already been called, in which case the
name fits.

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

Done.

+///
+/// This can be used when a sink is needed but no events are actually intended
+/// to be added. The [new ClosedStreamSink.error] constructor can be used to

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

If it's not intended to receive events, why not throw if it does (that is,
actually start in a closed state like after calling close)?

You can call close multiple times.

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

It's meant to be used to mimic the remote stream being closed—see for example
the use in StreamSinkCompleter, where making it throw errors would cause a bunch
of pending adds to fail. In general, the may not know not to add to the sink.

+/// represent errors in creating a sink.

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

How does errors in a done future represent errors in *creating* a sink?

Is the created sink intended to replace the sink that couldn't be created due to
an error, and the done future is the only place to return the error? (IOW:
sentence is not clear, an example may improve it).

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

From the StreamSink documentation: "If the StreamSink fails with an error [...]
the done future will complete with that error." ClosedStreamSink.error
represents a failed sink—it's just that the sink failed before it was even
constructed.

I'll see if I can clarify that, though.

+class ClosedStreamSink<T> implements StreamSink<T> {
+  final Future done;
+
+  /// Whether the sink has been closed.
+  ///
+  /// To be consistent with other stream sinks that have been closed remotely,

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

I would use the name "canceled" for what the receiving end does (like a stream
subscription's cancel and the StreamController's onCancel).

Still, this is describing the class in terms of what something else does instead
of what it does itself (ignore events).

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

I think changing the name and updating the class description made this clear, so
I'm just removing the paragraph.

+  /// this ignores events until [close] is called. After that, it throws
+  /// [StateError]s.
+  var _closed = false;
+
+  /// Creates a closed sink.
+  ///
+  /// If [done] is passed, it's used as the [Sink.done] future. Otherwise, a
+  /// completed future is used.
+  ClosedStreamSink({Future done}) : done = done ?? new Future.value();
+
+  /// Creates a closed sink whose [done] future emits [error].
+  ClosedStreamSink.error(error, [StackTrace stackTrace])
+      : done = new Future.error(error, stackTrace);

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

This will cause the error to be uncaught in the next microtask, which may be too
soon for the user of the sink to call close().
Consider ignoring the error:
  new Future.error(..,..)..catchError((_){});
or storing it separately until the user actually asks for a done future.

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

Done.

+
+  void add(T data) {
+    if (!_closed) return;
+    throw new StateError("Cannot add to a closed sink.");

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

How about just:
  if (_closed) throw ...;
The current code seems like a round-about way of doing that (as in: I had to
read it twice to figure out that that was all it did).

Consider a helper function to create the state error (to avoid duplicating the
message):

  if (_closed) throw _closedError();
...
_closedError => new StateError("...");

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

Done.

+  }
+
+  void addError(error, [StackTrace stackTrace]) {
+    if (!_closed) return;
+    throw new StateError("Cannot add to a closed sink.");
+  }
+
+  Future addStream(Stream<T> stream) {
+    if (_closed) {
+      throw new StateError("Cannot add to a closed sink.");
+    }
+

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

You need a bit here to remember that you are adding a stream.

You should get a state error if you try to add again (or close) until the
addStream has completed.

Maybe change _closed to a state which is one of OPEN, ADDING or CLOSED, so
add(..) becomes: if (_state == OPEN) return; if (_state == ADDING) throw
_addWhileAddingError(); assert(_state == CLOSED); throw _closedError();

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

Done.

+    return stream.listen(null).cancel() ?? new Future.value();

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

In dart:async we have a Future<dynamic> completed with null that we reuse
occasionally.
Do you think we should make that public so you could use it here instead of
creating a new future every time?

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

I don't think it's worth the complexity.

+  }
+
+  Future close() {
+    _closed = true;
+    return done;
+  }
+}