Added toMultiSubscriber on Stream.

sdk/lib/async/stream_pipe.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;
 
 /**
- * A pipe between two streams.
+ * A wrapper around a stream that allows independent subscribers.
  *
- * The default pipe subscribes to the [source] and sends on the
- * [stream].
+ * The default behavior is to subscribes to the [source] and sends on the

[floitsch, 2013/01/09 16:02:49]

By default [this] subscribes to [_source] and forwards all events to its own
subscribers.

[Lasse Reichstein Nielsen, 2013/01/09 16:16:43]

Done.

+ * this [Stream].
  *
  * The events are passed through the [_handleData], [_handleError] and
  * [_handleDone] methods. Subclasses are supposed to add handling of some of
  * the events by overriding these methods.
  *
- * This class is intended for internal use only. Users can use the [PipeStream]
- * to configure similar behavior.
+ * This class is intended for internal use only.
  */
-abstract class _ForwardingStream<S, T> extends _MultiStreamImpl<T>
-                                       implements StreamTransformer<S, T> {
+class _ForwardingMultiStream<S, T> extends _MultiStreamImpl<T> {
   Stream<S> _source = null;
   StreamSubscription _subscription = null;
 
   void _subscribeToSource() {
     _subscription = _source.listen(this._handleData,
                                    onError: this._handleError,
                                    onDone: this._handleDone);
     if (_isPaused) {
       _subscription.pause();
     }
   }
 
-  Stream<T> bind(Stream<S> source) {
-    assert(_source == null);
-    _source = source;
-    if (_hasSubscribers) {
-      _subscribeToSource();
-    }
-    return this;
-  }
-
   /**
     * Subscribe or unsubscribe on [source] depending on whether
     * [stream] has subscribers.
     */
   void _onSubscriptionStateChange() {
     if (_hasSubscribers) {
       assert(_subscription == null);
       if (_source != null) {
         _subscribeToSource();
       }
@@ skipping 22 matching lines @@
   void _handleError(AsyncError error) {
     _signalError(error);
   }
 
   void _handleDone() {
     _close();
   }
 }
 
 
+abstract class _ForwardingTransformer<S, T> extends _ForwardingMultiStream<S, T>
+                                            implements StreamTransformer<S, T> {
+  Stream<T> bind(Stream<S> source) {
+    assert(_source == null);
+    _source = source;
+    if (_hasSubscribers) {
+      _subscribeToSource();
+    }
+    return this;
+  }
+}
+
 // -------------------------------------------------------------------
-// Stream pipes used by the default Stream implementation.
+// Stream transformers used by the default Stream implementation.
 // -------------------------------------------------------------------
 
 typedef bool _Predicate<T>(T value);
 
-class WhereStream<T> extends _ForwardingStream<T, T> {
+class WhereStream<T> extends _ForwardingTransformer<T, T> {
   final _Predicate<T> _test;
 
   WhereStream(bool test(T value))
       : this._test = test;
 
   void _handleData(T inputEvent) {
     bool satisfies;
     try {
       satisfies = _test(inputEvent);
     } catch (e, s) {
       _signalError(new AsyncError(e, s));
       return;
     }
     if (satisfies) {
       _add(inputEvent);
     }
   }
 }
 
 
 typedef T _Transformation<S, T>(S value);
 
 /**
  * A stream pipe that converts data events before passing them on.
  */
-class MapStream<S, T> extends _ForwardingStream<S, T> {
+class MapStream<S, T> extends _ForwardingTransformer<S, T> {
   final _Transformation _transform;
 
   MapStream(T transform(S event))
       : this._transform = transform;
 
   void _handleData(S inputEvent) {
     T outputEvent;
     try {
       outputEvent = _transform(inputEvent);
     } catch (e, s) {
       _signalError(new AsyncError(e, s));
       return;
     }
     _add(outputEvent);
   }
 }
 
 /**
  * A stream pipe that converts data events before passing them on.
  */
-class ExpandStream<S, T> extends _ForwardingStream<S, T> {
+class ExpandStream<S, T> extends _ForwardingTransformer<S, T> {
   final _Transformation<S, Iterable<T>> _expand;
 
   ExpandStream(Iterable<T> expand(S event))
       : this._expand = expand;
 
   void _handleData(S inputEvent) {
     try {
       for (T value in _expand(inputEvent)) {
         _add(value);
       }
     } catch (e, s) {
       // If either _expand or iterating the generated iterator throws,
       // we abort the iteration.
       _signalError(new AsyncError(e, s));
     }
   }
 }
 
 
 typedef AsyncError _ErrorTransformation(AsyncError error);
 
 /**
  * A stream pipe that converts or disposes error events
  * before passing them on.
  */
-class HandleErrorStream<T> extends _ForwardingStream<T, T> {
+class HandleErrorStream<T> extends _ForwardingTransformer<T, T> {
   final _ErrorTransformation _transform;
 
   HandleErrorStream(AsyncError transform(AsyncError event))
       : this._transform = transform;
 
   void _handleError(AsyncError error) {
     try {
       error = _transform(error);
       if (error == null) return;
     } catch (e, s) {
@@ skipping 10 matching lines @@
 
 /**
  * A stream pipe that intercepts all events and can generate any event as
  * output.
  *
  * Each incoming event on this [StreamSink] is passed to the corresponding
  * provided event handler, along with a [StreamSink] linked to the [output] of
  * this pipe.
  * The handler can then decide which events to send to the output
  */
-class PipeStream<S, T> extends _ForwardingStream<S, T> {
+class PipeStream<S, T> extends _ForwardingTransformer<S, T> {
   final _TransformDataHandler<S, T> _onData;
   final _TransformErrorHandler<T> _onError;
   final _TransformDoneHandler<T> _onDone;
   StreamSink<T> _sink;
 
   PipeStream({void onData(S data, StreamSink<T> sink),
               void onError(AsyncError data, StreamSink<T> sink),
               void onDone(StreamSink<T> sink)})
       : this._onData = (onData == null ? _defaultHandleData : onData),
         this._onError = (onError == null ? _defaultHandleError : onError),
@@ skipping 51 matching lines @@
 
 /**
  * A stream pipe that intercepts all events and can generate any event as
  * output.
  *
  * Each incoming event on this [StreamSink] is passed to the corresponding
  * method on [transform], along with a [StreamSink] linked to the [output] of
  * this pipe.
  * The handler can then decide which events to send to the output
  */
-class TransformStream<S, T> extends _ForwardingStream<S, T> {
+class TransformStream<S, T> extends _ForwardingTransformer<S, T> {
   final StreamTransformer<S, T> _transform;
   StreamSink<T> _sink;
 
   TransformStream(StreamTransformer<S, T> transform)
       : this._transform = transform {
     // Cache the sink wrapper to avoid creating a new one for each event.
     this._sink = new _StreamImplSink(this);
   }
 
   void _handleData(S data) {
@@ skipping 45 matching lines @@
   void handleError(AsyncError error, StreamSink<T> sink) {
     _handleError(error, sink);
   }
 
   void handleDone(StreamSink<T> sink) {
     _handleDone(sink);
   }
 }
 
 
-class TakeStream<T> extends _ForwardingStream<T, T> {
+class TakeStream<T> extends _ForwardingTransformer<T, T> {
   int _remaining;
 
   TakeStream(int count)
       : this._remaining = count {
     if (count is! int) throw new ArgumentError(count);
   }
 
   void _handleData(T inputEvent) {
     if (_remaining > 0) {
       _add(inputEvent);
       _remaining -= 1;
       if (_remaining == 0) {
         // Closing also unsubscribes all subscribers, which unsubscribes
         // this from source.
         _close();
       }
     }
   }
 }
 
 
-class TakeWhileStream<T> extends _ForwardingStream<T, T> {
+class TakeWhileStream<T> extends _ForwardingTransformer<T, T> {
   final _Predicate<T> _test;
 
   TakeWhileStream(bool test(T value))
       : this._test = test;
 
   void _handleData(T inputEvent) {
     bool satisfies;
     try {
       satisfies = _test(inputEvent);
     } catch (e, s) {
       _signalError(new AsyncError(e, s));
       // The test didn't say true. Didn't say false either, but we stop anyway.
       _close();
       return;
     }
     if (satisfies) {
       _add(inputEvent);
     } else {
       _close();
     }
   }
 }
 
-class SkipStream<T> extends _ForwardingStream<T, T> {
+class SkipStream<T> extends _ForwardingTransformer<T, T> {
   int _remaining;
 
   SkipStream(int count)
       : this._remaining = count{
     if (count is! int) throw new ArgumentError(count);
   }
 
   void _handleData(T inputEvent) {
     if (_remaining > 0) {
       _remaining--;
       return;
     }
     return _add(inputEvent);
   }
 }
 
-class SkipWhileStream<T> extends _ForwardingStream<T, T> {
+class SkipWhileStream<T> extends _ForwardingTransformer<T, T> {
   final _Predicate<T> _test;
   bool _hasFailed = false;
 
   SkipWhileStream(bool test(T value))
       : this._test = test;
 
   void _handleData(T inputEvent) {
     if (_hasFailed) {
       _add(inputEvent);
     }
     bool satisfies;
     try {
       satisfies = _test(inputEvent);
     } catch (e, s) {
       _signalError(new AsyncError(e, s));
       // A failure to return a boolean is considered "not matching".
       _hasFailed = true;
       return;
     }
     if (!satisfies) {
       _hasFailed = true;
       _add(inputEvent);
     }
   }
 }
 
 typedef bool _Equality<T>(T a, T b);
 
-class DistinctStream<T> extends _ForwardingStream<T, T> {
+class DistinctStream<T> extends _ForwardingTransformer<T, T> {
   static var _SENTINEL = new Object();
 
   _Equality<T> _equals;
   var _previous = _SENTINEL;
 
   DistinctStream(bool equals(T a, T b))
       : _equals = equals;
 
   void _handleData(T inputEvent) {
     if (identical(_previous, _SENTINEL)) {
@@ skipping 11 matching lines @@
         _signalError(new AsyncError(e, s));
         return null;
       }
       if (!isEqual) {
         _add(inputEvent);
         _previous = inputEvent;
       }
     }
   }
 }