Move _StreamSinkImpl from dart:io to dart:async as StreamSinkAdapter.

sdk/lib/async/stream.dart

 // Copyright (c) 2013, 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.
 
 part of dart.async;
 
 // -------------------------------------------------------------------
 // Core Stream types
 // -------------------------------------------------------------------
 
@@ skipping 1308 matching lines @@
    *
    * * The synchronous methods of [EventSink] were called, resulting in an
    *   error. If there is no active future (like from an addStream call), the
    *   [done] future will complete with that error
    */
   Future get done;
 }
 
 
 /**
+ * A `StreamSinkAdapter` is an adapter implementation that can be used on a

[Lasse Reichstein Nielsen, 2014/03/18 15:34:45]

Too long first line. How about:

  An adapter that allows using a [StreamConsumer] as a [StreamSink].

Or perhaps:

  A [StreamSink] adapter for a [StreamConsumer].

(The word "adapter" should be fine, since it is a classical OO pattern).

[Anders Johnsen, 2014/03/19 11:48:55]

Done.

+ * [StreamConsumer], to facilitate the functionality of a [StreamSink],
+ *
+ * The `StreamSinkAdapter` buffers the input given by all [EventSink] methods
+ * and will delay an [addStream] until the buffer is flushed.

[Lasse Reichstein Nielsen, 2014/03/18 15:34:45]

the buffer -> that buffer

This sounds like nothing is written until you call "flush"?
How about:  
  ... until all buffered data has been forwarded to the stream consumer.

[Anders Johnsen, 2014/03/19 11:48:55]

Done.

+ *
+ * When the `StreamSinkAdapter` is bound to a stream (through [addStream]) any

[Lasse Reichstein Nielsen, 2014/03/18 15:34:45]

When -> While

[Anders Johnsen, 2014/03/19 11:48:55]

Done.

+ * call to the `StreamSinkAdapter` will throw a [StateError].

[Lasse Reichstein Nielsen, 2014/03/18 15:34:45]

Is this all calls? What about "done" (which is a getter, not a method, but
still...)?

[Anders Johnsen, 2014/03/19 11:48:55]

Yes, only done... :)

+ * When the [addStream] completes, the `StreamSinkAdapter` will again be open
+ * to all calls.
+ *
+ * If events are added to the `StreamSinkAdapter` after the adapter is closed,
+ * the events will be ignored.
+ * Use the [done] future to be notified when the `StreamSinkAdapter` is closed.
+ */
+class StreamSinkAdapter<S> implements StreamSink<S> {
+  final StreamConsumer<S> _target;
+  Completer _doneCompleter = new Completer();
+  Future _doneFuture;
+  StreamController<S> _controllerInstance;
+  Completer _controllerCompleter;
+  bool _isClosed = false;
+  bool _isBound = false;
+  bool _hasError = false;
+
+  /**
+   * Construct a new StreamSinkAdapter, from a `target` [StreamConsumer].
+   */
+  StreamSinkAdapter(StreamConsumer<S> target) : _target = target {
+    _doneFuture = _doneCompleter.future;
+  }
+
+  void add(S data) {

[Lasse Reichstein Nielsen, 2014/03/18 15:34:45]

No _isBound check?

[Anders Johnsen, 2014/03/19 11:48:55]

Is in _controller getter.

+    if (_isClosed) return;
+    _controller.add(data);
+  }
+
+  void addError(error, [StackTrace stackTrace]) =>
+      _controller.addError(error, stackTrace);

[Lasse Reichstein Nielsen, 2014/03/18 15:34:45]

No _isBound check?

[Anders Johnsen, 2014/03/19 11:48:55]

Is in _controller getter.

+
+  Future addStream(Stream<S> stream) {
+    if (_isBound) {
+      throw new StateError("StreamSink is already bound to a stream");
+    }
+    _isBound = true;
+    if (_hasError) return done;
+    // Wait for any sync operations to complete.
+    Future targetAddStream() {
+      return _target.addStream(stream)
+          .whenComplete(() {
+            _isBound = false;
+          });
+    }
+    if (_controllerInstance == null) return targetAddStream();
+    var future = _controllerCompleter.future;
+    _controllerInstance.close();
+    return future.then((_) => targetAddStream());
+  }
+
+  /**
+   * Returns a [Future] that completes once all buffered events is accepted by
+   * the underlying [StreamConsumer].
+   *
+   * It's an error to call this method, while an [addStream] is incomplete.
+   *
+   * NOTE: This is not necessarily the same as the data being flushed by the

[Lasse Reichstein Nielsen, 2014/03/18 15:34:45]

This note is confusion. It says what this call isn't (or might not be), which is
extremely vague. Is it sometimes the same? When? What is it when it is not the
same?

How about:
NOTE: This method does not guarantee anything except that the stream consumer
has received the data. It does not guarantee that the consumer has acted on the
data in any way, or that the data has reached its final destination.

[Anders Johnsen, 2014/03/19 11:48:55]

Done.

+   * operating system.
+   */
+  Future flush() {
+    if (_isBound) {

[Lasse Reichstein Nielsen, 2014/03/18 15:34:45]

Consider having a helper function for this check, e.g., "_checkNotBound()".

[Anders Johnsen, 2014/03/19 11:48:55]

Done.

+      throw new StateError("StreamSink is bound to a stream");
+    }
+    if (_controllerInstance == null) return new Future.value(this);
+    // Adding an empty stream-controller will return a future that will complete
+    // when all data is done.
+    _isBound = true;
+    var future = _controllerCompleter.future;
+    _controllerInstance.close();
+    return future.whenComplete(() {
+          _isBound = false;
+        });
+  }
+
+  Future close() {
+    if (_isBound) {
+      throw new StateError("StreamSink is bound to a stream");
+    }
+    if (!_isClosed) {
+      _isClosed = true;
+      if (_controllerInstance != null) {
+        _controllerInstance.close();
+      } else {
+        _closeTarget();
+      }
+    }
+    return done;
+  }
+
+  Future get done => _doneFuture;
+
+  // Private helper methods.
+
+  void _closeTarget() {
+    _target.close()
+        .then((value) => _completeDone(value: value),
+              onError: (error) => _completeDone(error: error));
+  }
+
+  void _completeDone({value, error}) {
+    if (_doneCompleter == null) return;
+    if (error == null) {
+      _doneCompleter.complete(value);
+    } else {
+      _hasError = true;
+      _doneCompleter.completeError(error);
+    }
+    _doneCompleter = null;
+  }
+
+  StreamController<S> get _controller {
+    if (_isBound) {
+      throw new StateError("StreamSink is bound to a stream");
+    }
+    if (_isClosed) {
+      throw new StateError("StreamSink is closed");
+    }
+    if (_controllerInstance == null) {
+      _controllerInstance = new StreamController<S>(sync: true);
+      _controllerCompleter = new Completer();
+      _target.addStream(_controller.stream)
+          .then(
+              (_) {
+                if (_isBound) {
+                  // A new stream takes over - forward values to that stream.
+                  _controllerCompleter.complete(this);
+                  _controllerCompleter = null;
+                  _controllerInstance = null;
+                } else {
+                  // No new stream, .close was called. Close _target.
+                  _closeTarget();
+                }
+              },
+              onError: (error) {
+                if (_isBound) {
+                  // A new stream takes over - forward errors to that stream.
+                  _controllerCompleter.completeError(error);
+                  _controllerCompleter = null;
+                  _controllerInstance = null;
+                } else {
+                  // No new stream. No need to close target, as it have already
+                  // failed.
+                  _completeDone(error: error);
+                }
+              });
+    }
+    return _controllerInstance;
+  }
+}
+
+
+/**
  * The target of a [Stream.transform] call.
  *
  * The [Stream.transform] call will pass itself to this object and then return
  * the resulting stream.
  *
  * It is good practice to write transformers that can be used multiple times.
  */
 abstract class StreamTransformer<S, T> {
 
   /**
@@ skipping 143 matching lines @@
 class _ControllerEventSinkWrapper<T> implements EventSink<T> {
   EventSink _sink;
   _ControllerEventSinkWrapper(this._sink);
 
   void add(T data) { _sink.add(data); }
   void addError(error, [StackTrace stackTrace]) {
     _sink.addError(error, stackTrace);
   }
   void close() { _sink.close(); }
 }