Make Stream transformation respect the single/multi subscriber nature of the source.

sdk/lib/async/stream.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;
 
 // -------------------------------------------------------------------
 // Core Stream types
 // -------------------------------------------------------------------
 
@@ skipping 13 matching lines @@
     return stream;
   }
 
   /**
    * Creates a single-subscription stream that gets its data from [data].
    */
   factory Stream.fromIterable(Iterable<T> data) {
     return new _IterableSingleStreamImpl<T>(data);
   }
 
+  /** Whether the stream is a single-subscription stream. */
+  bool get isSingleSubscription;

[floitsch, 2013/01/14 16:09:09]

different CL, but sometimes we have singleSubscriber, and sometimes
singleSubscription. We should be consistent.

+
   /**
    * Returns a multi-subscription stream that produces the same events as this.
    *
    * If this stream is single-subscription, return a new stream that allows
    * multiple subscribers. It will subscribe to this stream when its first
    * subscriber is added, and unsubscribe again when the last subscription is
    * cancelled.
    *
    * If this stream is already multi-subscriber, it is returned unmodified.
    */
@@ skipping 45 matching lines @@
                                  void onDone(),
                                  bool unsubscribeOnError});
 
   /**
    * Creates a new stream from this stream that discards some data events.
    *
    * The new stream sends the same error and done events as this stream,
    * but it only sends the data events that satisfy the [test].
    */
   Stream<T> where(bool test(T event)) {
-    return this.transform(new WhereStream<T>(test));
+    return this.transform(new WhereTransformer<T>(test));
   }
 
   /**
    * Create a new stream that converts each element of this stream
    * to a new value using the [convert] function.
    */
   Stream mappedBy(convert(T event)) {
-    return this.transform(new MapStream<T, dynamic>(convert));
+    return this.transform(new MapTransformer<T, dynamic>(convert));
   }
 
   /**
    * Create a wrapper Stream that intercepts some errors from this stream.
    *
    * If this stream sends an error that matches [test], then it is intercepted
    * by the [handle] function.
    *
    * An [AsyncError] [:e:] is matched by a test function if [:test(e):] returns
    * true. If [test] is omitted, every error is considered matching.
    *
    * If the error is intercepted, the [handle] function can decide what to do
    * with it. It can throw if it wants to raise a new (or the same) error,
    * or simply return to make the stream forget the error.

[floitsch, 2013/01/14 16:09:09]

different CL: "if the error needs to be transformed into a value, use the [XYZ]
transformer."
Maybe just add a TODO for now?.

[Lasse Reichstein Nielsen, 2013/01/15 07:19:51]

Done.

    */
   Stream<T> handleError(void handle(AsyncError error), { bool test(error) }) {
-    return this.transform(new HandleErrorStream<T>(handle, test));
+    return this.transform(new HandleErrorTransformer<T>(handle, test));
   }
 
   /**
    * Create a new stream from this stream that converts each element
    * into zero or more events.
    *
    * Each incoming event is converted to an [Iterable] of new events,
    * and each of these new events are then sent by the returned stream
    * in order.
    */
   Stream expand(Iterable convert(T value)) {
-    return this.transform(new ExpandStream<T, dynamic>(convert));
+    return this.transform(
+        new ExpandTransformer<T, dynamic>(convert));
   }
 
   /**
    * Bind this stream as the input of the provided [StreamConsumer].
    */
   Future pipe(StreamConsumer<dynamic, T> streamConsumer) {
     return streamConsumer.consume(this);
   }
 
   /**
@@ skipping 281 matching lines @@
   /**
    * Provide at most the first [n] values of this stream.
    *
    * Forwards the first [n] data events of this stream, and all error
    * events, to the returned stream, and ends with a done event.
    *
    * If this stream produces fewer than [count] values before it's done,
    * so will the returned stream.
    */
   Stream<T> take(int count) {
-    return this.transform(new TakeStream(count));
+    return this.transform(new TakeTransformer<T>(count));
   }
 
   /**
    * Forwards data events while [test] is successful.
    *
    * The returned stream provides the same events as this stream as long
    * as [test] returns [:true:] for the event data. The stream is done
    * when either this stream is done, or when this stream first provides
    * a value that [test] doesn't accept.
    */
   Stream<T> takeWhile(bool test(T value)) {
-    return this.transform(new TakeWhileStream(test));
+    return this.transform(new TakeWhileTransformer<T>(test));
   }
 
   /**
    * Skips the first [count] data events from this stream.
    */
   Stream<T> skip(int count) {
-    return this.transform(new SkipStream(count));
+    return this.transform(new SkipTransformer<T>(count));
   }
 
   /**
    * Skip data events from this stream while they are matched by [test].
    *
    * Error and done events are provided by the returned stream unmodified.
    *
    * Starting with the first data event where [test] returns true for the
    * event data, the returned stream will have the same events as this stream.
    */
   Stream<T> skipWhile(bool test(T value)) {
-    return this.transform(new SkipWhileStream(test));
+    return this.transform(new SkipWhileTransformer<T>(test));
   }
 
   /**
    * Skip data events if they are equal to the previous data event.
    *
    * The returned stream provides the same events as this stream, except
    * that it never provides two consequtive data events that are equal.
    *
    * Equality is determined by the provided [equals] method. If that is
    * omitted, the '==' operator on the last provided data element is used.
    */
   Stream<T> distinct([bool equals(T previous, T next)]) {
-    return this.transform(new DistinctStream(equals));
+    return this.transform(new DistinctTransformer<T>(equals));
   }
 
   /**
    * Returns the first element.
    *
    * If [this] is empty throws a [StateError]. Otherwise this method is
    * equivalent to [:this.elementAt(0):]
    */
   Future<T> get first {
-    _FutureImpl<T> future = new _FutureImpl();
+    _FutureImpl<T> future = new _FutureImpl<T>();
     StreamSubscription subscription;
     subscription = this.listen(
       (T value) {
         future._setValue(value);
         subscription.cancel();
         return;
       },
       onError: future._setError,
       onDone: () {
         future._setError(new AsyncError(new StateError("No elements")));
@@ skipping 194 matching lines @@
   /**
    * Returns the value of the [index]th data event of this stream.
    *
    * If an error event occurs, the future will end with this error.
    *
    * If this stream provides fewer than [index] elements before closing,
    * an error is reported.
    */
   Future<T> elementAt(int index) {
     if (index is! int || index < 0) throw new ArgumentError(index);
-    _FutureImpl<T> future = new _FutureImpl();
+    _FutureImpl<T> future = new _FutureImpl<T>();
     StreamSubscription subscription;
     subscription = this.listen(
       (T value) {
         if (index == 0) {
           future._setValue(value);
           subscription.cancel();
           return;
         }
         index -= 1;
       },
@@ skipping 61 matching lines @@
   void signalError(AsyncError errorEvent);
   void close();
 }
 
 /** [Stream] wrapper that only exposes the [Stream] interface. */
 class StreamView<T> extends Stream<T> {
   Stream<T> _stream;
 
   StreamView(this._stream);
 
+  bool get isSingleSubscription => _stream.isSingleSubscription;
+
   StreamSubscription<T> listen(void onData(T value),
                                { void onError(AsyncError error),
                                  void onDone(),
                                  bool unsubscribeOnError }) {
     return _stream.listen(onData, onError: onError, onDone: onDone,
                           unsubscribeOnError: unsubscribeOnError);
   }
 }
 
 /**
@@ skipping 32 matching lines @@
    * Create a [StreamTransformer] that delegates events to the given functions.
    *
    * If a parameter is omitted, a default handler is used that forwards the
    * event directly to the sink.
    *
    * Pauses on the are forwarded to the input stream as well.
    */
   factory StreamTransformer.from({
       void onData(S data, StreamSink<T> sink),
       void onError(AsyncError error, StreamSink<T> sink),
-      void onDone(StreamSink<T> sink)}) = _StreamTransformerFunctionWrapper;
+      void onDone(StreamSink<T> sink)}) {
+    return new _StreamTransformerImpl<S, T>(onData, onError, onDone);
+  }
 
   Stream<T> bind(Stream<S> stream);
 }
 
 
 // TODO(lrn): Remove this class.
 /**
  * A base class for configuration objects for [TransformStream].
  *
  * A default implementation forwards all incoming events to the output sink.
@@ skipping 20 matching lines @@
     sink.signalError(error);
   }
 
   /**
    * Handle an incoming done event.
    */
   void handleDone(StreamSink<T> sink) {
     sink.close();
   }
 }