Add StreamQueue.fork and ForkableStream.

lib/src/stream_queue.dart

Index: lib/src/stream_queue.dart
diff --git a/lib/src/stream_queue.dart b/lib/src/stream_queue.dart
index 7d78ac593c79181bb504897f8f3b611130a3b48b..ba4041eefaa1014cead7eea1a5f56be503b0204b 100644
--- a/lib/src/stream_queue.dart
+++ b/lib/src/stream_queue.dart
@@ -7,6 +7,7 @@ library async.stream_events;
 import 'dart:async';
 import 'dart:collection';
 
+import "forkable_stream.dart";
 import "subscription_stream.dart";
 import "stream_completer.dart";
 import "../result.dart";
@@ -78,7 +79,7 @@ class StreamQueue<T> {
   // by the content of the fifth event.
 
   /// Source of events.
-  final Stream _sourceStream;
+  final ForkableStream _sourceStream;
 
   /// Subscription on [_sourceStream] while listening for events.
   ///
@@ -104,7 +105,9 @@ class StreamQueue<T> {
 
   /// Create a `StreamQueue` of the events of [source].
   StreamQueue(Stream source)
-      : _sourceStream = source;
+      : _sourceStream = source is ForkableStream
+          ? source
+          : new ForkableStream(source);

[Lasse Reichstein Nielsen, 2015/07/15 20:10:51]

Would it be possible to make the fork method without it costing when you don't
use it?
Maybe after calling fork, all events are forwarded to a different controller as
well as being put in the event queue, but if you don't fork, you won't create
the forkable stream.

Also, if you do "rest" as the first thing, I think we return the original
stream, which would now be a ForkableStream.

[nweiz, 2015/07/15 22:19:43]

I don't think this is possible without drastically complicating the
implementation. The reason I created ForkableStream in the first place was
because the complexity overhead of trying to manage a bunch of fork information
in StreamQueue, including pausing and closing the underlying stream at the right
time, became unmanageable.

Hopefully forwarding via a single synchronous StreamController isn't that much
overhead, though, since that's a common pattern across most higher-order stream
manipulation.

[Lasse Reichstein Nielsen, 2015/07/16 14:01:23]

I still think this could/should be handled at a different level than wrapping
the original stream.

For example you could "fork" the subscription when needed, into two
subscriptions providing the same values, and with individual buffering and
shared pause if all are paused. Then you could replace the original subscription
with one fork and convert the other fork to a stream using SubscriptionStream.

Or have a ForkableSubscription wrapper that you apply the first time you fork
the stream queue, then you can do further forks on that.
That way you would still only pay if you use it.

See example: https://codereview.chromium.org/1238503004/

When the stream is active, it's actually the subscription object that's the
active part, the Stream object can be discarded.

[nweiz, 2015/07/17 20:30:16]

In your example, just forking a stream causes it to listen, which is a
fundamental problem with dealing with subscriptions—they only exist when a
stream has been listened to. The underlying stream should only be listened to
once someone has actually requested a value from it; that shouldn't change based
on how many forks exist.

 
   /// Asks if the stream has any more events.
   ///
@@ -216,6 +219,22 @@ class StreamQueue<T> {
     throw _failClosed();
   }
 
+  /// Creates a new stream queue in the same position as this one.
+  ///
+  /// The fork is subscribed to the same underlying stream as this queue, but
+  /// it's otherwise wholly independent. If requests are made on one, they don't
+  /// move the other forward; if one is closed, the other is still open.
+  ///
+  /// The underlying stream will only be paused when all forks have no
+  /// outstanding requests, and only canceled when all forks are canceled.
+  StreamQueue<T> fork() {

[Lasse Reichstein Nielsen, 2015/07/16 14:01:23]

I have another problem with fork: It may cause buffering.

The StreamQueue is desinged so that it reads events until it can complete a
request. It never buffers events that it doesn't need.

When you fork, you have two independent consumers of the same events, so unless
they are in perfect sync, one of them will have to buffer events while its
paused - you add events to a paused completer beause you have nowhere else to
put them.
So if you fork and don't use the forked queue, it will accumulate events
forever. 
I don't like it when that can happen due to an innocuously looking API function
- it should be something you opt in to deliberately, not by accident.

For that reason, I'd prefer to not have fork at all.

[nweiz, 2015/07/17 20:30:16]

I can't emphasize enough how important forking is for creating higher-order
operations over stream queues, or how important those higher-order operations
are for testing streams. Right now we have *no* standard facilities for making
assertions against streams other than converting them to lists, which obviously
doesn't work for things like a process's standard output. The closest thing we
have—what pub uses—is scheduled_test's ScheduledStream and stream matchers,
which are essentially the same idea as StreamQueue. These actually work really
well, and there's one thing at the heart that makes that possible: forking.
Without fork(), we couldn't write matchers with the flexibility necessary to
handle the heterogeneous output that crops up in real-world applications.

I also think it's pretty clear when buffering will happen—users expect there to
be buffering somewhere when there's an unlistened single-subscription stream,
and the same goes for a StreamQueue. It's also pretty likely that even a single
stream queue will have some buffering when it's paused, and there's even less
visibility into that. Ultimately—and I say this as one of the heaviest users of
Dart streams in practice—I don't think that buffering is a huge concern for
users as long as it's reasonably straightforward to track down where it's
happening and fix it if it becomes a problem.

+    if (_isClosed) throw _failClosed();
+
+    var request = new _ForkRequest<T>(this);
+    _addRequest(request);
+    return request.queue;
+  }
+
   /// Cancels the underlying stream subscription.
   ///
   /// If [immediate] is `false` (the default), the cancel operation waits until
@@ -236,14 +255,15 @@ class StreamQueue<T> {
     if (_isClosed) throw _failClosed();
     _isClosed = true;
 
+    if (_isDone) return new Future.value();
+    if (_subscription == null) _subscription = _sourceStream.listen(null);
+
     if (!immediate) {
       var request = new _CancelRequest(this);
       _addRequest(request);
       return request.future;
     }
 
-    if (_isDone) return new Future.value();
-    if (_subscription == null) _subscription = _sourceStream.listen(null);
     var future = _subscription.cancel();
     _onDone();
     return future;
@@ -333,6 +353,7 @@ class StreamQueue<T> {
         return;
       }
     }
+
     if (!_isDone) {
       _subscription.pause();
     }
@@ -627,3 +648,50 @@ class _HasNextRequest<T> implements _EventRequest {
     _completer.complete(false);
   }
 }
+
+/// Request for a [StreamQueue.fork] call.
+class _ForkRequest<T> implements _EventRequest {
+  /// Completer for the stream used by the queue by the `fork` call.
+  StreamCompleter _completer;
+
+  StreamQueue<T> queue;
+
+  /// The [StreamQueue] object that has this request queued.
+  final StreamQueue _streamQueue;
+
+  _ForkRequest(this._streamQueue) {
+    _completer = new StreamCompleter<T>();
+    queue = new StreamQueue<T>(_completer.stream);
+  }
+
+  bool addEvents(Queue<Result> events) {
+    _completeStream(events);
+    return true;
+  }
+
+  void close(Queue<Result> events) {
+    _completeStream(events);
+  }
+
+  void _completeStream(Queue<Result> events) {
+    if (events.isEmpty) {
+      if (_streamQueue._isDone) {
+        _completer.setEmpty();
+      } else {
+        _completer.setSourceStream(_streamQueue._sourceStream.fork());
+      }
+    } else {
+      // There are prefetched events which need to be added before the
+      // remaining stream.
+      var controller = new StreamController<T>();
+      for (var event in events) {
+        event.addTo(controller);
+      }
+
+      var fork = _streamQueue._sourceStream.fork();
+      controller.addStream(fork, cancelOnError: false)
+          .whenComplete(controller.close);
+      _completer.setSourceStream(controller.stream);
+    }
+  }
+}