Warn when adding something to a closed sink and improve documentation

sdk/lib/async/stream_transformers.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;
 
 /**
  * Wraps an [_EventSink] so it exposes only the [EventSink] interface.
  */
 class _EventSinkWrapper<T> implements EventSink<T> {
@@ skipping 191 matching lines @@
  * Wraps handlers (from [StreamTransformer.fromHandlers]) into an `EventSink`.
  *
  * This way we can reuse the code from [_StreamSinkTransformer].
  */
 class _HandlerEventSink<S, T> implements EventSink<S> {
   final _TransformDataHandler<S, T> _handleData;
   final _TransformErrorHandler<T> _handleError;
   final _TransformDoneHandler<T> _handleDone;
 
   /// The output sink where the handlers should send their data into.
-  final EventSink<T> _sink;
+  EventSink<T> _sink;
 
   _HandlerEventSink(
-      this._handleData, this._handleError, this._handleDone, this._sink);
+      this._handleData, this._handleError, this._handleDone, this._sink) {
+    if (_sink == null) {
+      throw new ArgumentError("The provided sink must not be null.");
+    }
+  }
+
+  bool get _isClosed => _sink == null;
+
+  _reportClosedSink() {
+    // TODO(29554): throw a StateError, and don't just report the problem.
+    Zone.ROOT
+      ..print("Sink is closed and adding to it is an error.")
+      ..print("  See http://dartbug.com/29554.")
+      ..print(StackTrace.current.toString());
+  }
 
   void add(S data) {
+    if (_isClosed) {
+      _reportClosedSink();
+    }
     if (_handleData != null) {
       _handleData(data, _sink);
     } else {
       _sink.add(data as T);
     }
   }
 
   void addError(Object error, [StackTrace stackTrace]) {
+    if (_isClosed) {
+      _reportClosedSink();
+    }
     if (_handleError != null) {
       _handleError(error, stackTrace, _sink);
     } else {
       _sink.addError(error, stackTrace);
     }
   }
 
   void close() {
+    if (_isClosed) return;
+    var sink = _sink;
+    _sink = null;
     if (_handleDone != null) {
-      _handleDone(_sink);
+      _handleDone(sink);
     } else {
-      _sink.close();
+      sink.close();
     }
   }
 }
 
 /**
  * A StreamTransformer that transformers events with the given handlers.
  *
  * Note that this transformer can only be used once.
  */
 class _StreamHandlerTransformer<S, T> extends _StreamSinkTransformer<S, T> {
@@ skipping 52 matching lines @@
   StreamSubscription<T> listen(void onData(T event),
       {Function onError, void onDone(), bool cancelOnError}) {
     cancelOnError = identical(true, cancelOnError);
     StreamSubscription<T> result = _onListen(_stream, cancelOnError);
     result.onData(onData);
     result.onError(onError);
     result.onDone(onDone);
     return result;
   }
 }