Add StreamQueue.fork and ForkableStream.

lib/src/stream_queue.dart

 // Copyright (c) 2015, 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.
 
 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";
 
 /// An asynchronous pull-based interface for accessing stream events.
 ///
 /// Wraps a stream and makes individual events available on request.
 ///
 /// You can request (and reserve) one or more events from the stream,
 /// and after all previous requests have been fulfilled, stream events
@@ skipping 51 matching lines @@
   // again when new events are available.
   // The request can remove events that it uses, or keep them in the event
   // queue until it has all that it needs.
   //
   // This model is very flexible and easily extensible.
   // It allows requests that don't consume events (like [hasNext]) or
   // potentially a request that takes either five or zero events, determined
   // by the content of the fifth event.
 
   /// Source of events.
-  final Stream _sourceStream;
+  final ForkableStream _sourceStream;
 
   /// Subscription on [_sourceStream] while listening for events.
   ///
   /// Set to subscription when listening, and set to `null` when the
   /// subscription is done (and [_isDone] is set to true).
   StreamSubscription _subscription;
 
   /// Whether we have listened on [_sourceStream] and the subscription is done.
   bool _isDone = false;
 
   /// Whether a closing operation has been performed on the stream queue.
   ///
   /// Closing operations are [cancel] and [rest].
   bool _isClosed = false;
 
   /// Queue of events not used by a request yet.
   final Queue<Result> _eventQueue = new Queue();
 
   /// Queue of pending requests.
   ///
   /// Access through methods below to ensure consistency.
   final Queue<_EventRequest> _requestQueue = new Queue();
 
   /// 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.
   ///
   /// Returns a future that completes with `true` if the stream has any
   /// more events, whether data or error.
   /// If the stream closes without producing any more events, the returned
   /// future completes with `false`.
   ///
   /// Can be used before using [next] to avoid getting an error in the
   /// future returned by `next` in the case where there are no more events.
@@ skipping 91 matching lines @@
   Future<List<T>> take(int count) {
     if (count < 0) throw new RangeError.range(count, 0, null, "count");
     if (!_isClosed) {
       var request = new _TakeRequest<T>(count);
       _addRequest(request);
       return request.future;
     }
     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
   /// all previously requested events have been processed, then it cancels the
   /// subscription providing the events.
   ///
   /// If [immediate] is `true`, the subscription is instead canceled
   /// immediately. Any pending events are completed as though the underlying
   /// stream had closed.
   ///
   /// The returned future completes with the result of calling
   /// `cancel`.
   ///
   /// After calling `cancel`, no further events can be requested.
   /// None of [next], [rest], [skip], [take] or [cancel] may be
   /// called again.
   Future cancel({bool immediate: false}) {
     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;
   }
 
   /// Returns an error for when a request is made after cancel.
   ///
   /// Returns a [StateError] with a message saying that either
   /// [cancel] or [rest] have already been called.
   Error _failClosed() {
@@ skipping 69 matching lines @@
   ///
   /// Called after receiving an event.
   void _checkQueues() {
     while (_requestQueue.isNotEmpty) {
       if (_requestQueue.first.addEvents(_eventQueue)) {
         _requestQueue.removeFirst();
       } else {
         return;
       }
     }
+
     if (!_isDone) {
       _subscription.pause();
     }
   }
 
   /// Extracts the subscription and makes this stream queue unusable.
   ///
   /// Can only be used by the very last request.
   StreamSubscription _dispose() {
     assert(_isClosed);
@@ skipping 274 matching lines @@
       _completer.complete(true);
       return true;
     }
     return false;
   }
 
   void close(_) {
     _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);
+    }
+  }
+}