Add public-facing method and class that allows intercepting stream events.

sdk/lib/async/stream_impl.dart

 // Copyright (c) 2012, 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;
 
 // States shared by single/multi stream implementations.
 
 /// Initial and default state where the stream can receive and send events.
 const int _STREAM_OPEN = 0;
@@ skipping 44 matching lines @@
    * If events are added to the stream (using [_add], [_signalError] or [_done])
    * while the stream is paused, or while another event is firing, events will
    * stored here.
    * Also supports scheduling the events for later execution.
    */
   _PendingEvents _pendingEvents;
 
   // ------------------------------------------------------------------
   // Stream interface.
 
-  StreamSubscription listen(void onData(T data),
-                            { void onError(AsyncError error),
-                              void onDone(),
-                              bool unsubscribeOnError }) {
+  StreamSubscription<T> listen(void onData(T data),
+                               { void onError(AsyncError error),
+                                 void onDone(),
+                                 bool unsubscribeOnError }) {
     if (_isComplete) {
       return new _DoneSubscription(onDone);
     }
     if (onData == null) onData = _nullDataHandler;
     if (onError == null) onError = _nullErrorHandler;
     if (onDone == null) onDone = _nullDoneHandler;
     unsubscribeOnError = identical(true, unsubscribeOnError);
-    _StreamListener subscription =
+    _StreamSubscriptionImpl subscription =
         _createSubscription(onData, onError, onDone, unsubscribeOnError);
     _addListener(subscription);
     return subscription;
   }
 
   // ------------------------------------------------------------------
   // StreamSink interface-like methods for sending events into the stream.
   // It's the responsibility of the caller to ensure that the stream is not
   // paused when adding events. If the stream is paused, the events will be
   // queued, but it's better to not send events at all.
@@ skipping 998 matching lines @@
   void _delayDone() {
     assert(_timer == null && _pauseCount == 0);
     _timer = new Timer(0, (_) {
       if (_handler != null) _handler();
     });
   }
 
   bool get _isComplete => _timer == null && _pauseCount == 0;
 
   void onData(void handleAction(T value)) {}
-  void onError(void handleError(StateError error)) {}
-  void onDone(void handleDone(T value)) {
+
+  void onError(void handleError(AsyncError error)) {}
+
+  void onDone(void handleDone()) {
     _handler = handleDone;
   }
 
   void pause([Future signal]) {
     if (_isComplete) {
       throw new StateError("Subscription has been canceled.");
     }
     if (_timer != null) _timer.cancel();
     _pauseCount++;
   }
@@ skipping 52 matching lines @@
                                      onError: this._signalError,
                                      onDone: this._close);
     } else {
       // TODO(lrn): Check why this can happen.
       if (_subscription == null) return;
       _subscription.cancel();
       _subscription = null;
     }
   }
 }