Make it possible to not complete a request when the event source is done.

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 "subscription_stream.dart";
@@ skipping 42 matching lines @@
 ///       var headerCount =
 ///           first.parseInt(first.substring(MAGIC_MARKER.length + 1));
 ///       handleMessage(headers: await events.take(headerCount),
 ///                     body: events.rest);
 ///       return;
 ///     }
 ///     // Error handling.
 ///
 /// When you need no further events the `StreamQueue` should be closed
 /// using [cancel]. This releases the underlying stream subscription.
-class StreamQueue<T> {
+abstract class StreamQueue<T> {
   // This class maintains two queues: one of events and one of requests.
   // The active request (the one in front of the queue) is called with
   // the current event queue when it becomes active.
   //
   // If the request returns true, it's complete and will be removed from the
   // request queue.
   // If the request returns false, it needs more events, and will be called
   // 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;
-
-  /// 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.
+  /// Whether the event source done.

[nweiz, 2015/08/21 19:43:57]

"is done"

[Lasse Reichstein Nielsen, 2015/08/24 08:31:33]

Done.
I mean: 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;
+  factory StreamQueue(Stream source) = _StreamQueue<T>;
+
+  StreamQueue._();
 
   /// 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.
+  /// Another alternative is to use `take(1)` which returns either zero or
+  /// one events.
   Future<bool> get hasNext {
     if (!_isClosed) {
       var hasNextRequest = new _HasNextRequest();
       _addRequest(hasNextRequest);
       return hasNextRequest.future;
     }
     throw _failClosed();
   }
 
   /// Requests the next (yet unrequested) event from the stream.
@@ skipping 81 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();
   }
 
-  /// Cancels the underlying stream subscription.
+  /// Cancels the underlying event source.
   ///
   /// 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
+  /// If [immediate] is `true`, the source 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 (!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;
+    if (_isDone && _eventQueue.isEmpty) return new Future.value();
+    return _cancel();
   }
 
+  /// Cancels the underlying event source.
+  Future _cancel();

[nweiz, 2015/08/21 19:43:57]

It might be clearer to explicitly divide these private methods into two groups:
methods called by the subclass to signal something for the superclass, and
methods called by the superclass to signal something for the subclass. Without
that, I found it hard to parse out what's called when and in which direction.

[Lasse Reichstein Nielsen, 2015/08/24 08:31:34]

Good idea.

+
   /// 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() {
     return new StateError("Already cancelled");
   }
 
   // Callbacks receiving the events of the source stream.
-

[nweiz, 2015/08/21 19:43:57]

Nit: keep this empty line.

[Lasse Reichstein Nielsen, 2015/08/24 08:31:34]

I've documented the function instead.

-  void _onData(T data) {
-    _eventQueue.add(new Result.value(data));
-    _checkQueues();
-  }
-
-  void _onError(error, StackTrace stack) {
-    _eventQueue.add(new Result.error(error, stack));
+  void _addResult(Result result) {
+    _eventQueue.add(result);
     _checkQueues();
   }
 
   void _onDone() {

[nweiz, 2015/08/21 19:43:57]

Document this.

[Lasse Reichstein Nielsen, 2015/08/24 08:31:33]

Done.

-    _subscription = null;
     _isDone = true;
-    _closeAllRequests();
+    _checkQueues();
   }
 
   // Request queue management.
 
   /// Adds a new request to the queue.
   void _addRequest(_EventRequest request) {
-    if (_isDone) {
-      assert(_requestQueue.isEmpty);
-      if (!request.addEvents(_eventQueue)) {
-        request.close(_eventQueue);
-      }
-      return;
-    }
     if (_requestQueue.isEmpty) {
-      if (request.addEvents(_eventQueue)) return;
+      if (request.addEvents(_eventQueue, _isDone)) return;
       _ensureListening();
     }
     _requestQueue.add(request);
   }
 
-  /// Ensures that we are listening on events from [_sourceStream].
+  /// Ensures that we are listening on events from the event source.
   ///
-  /// Resumes subscription on [_sourceStream], or creates it if necessary.
+  /// Starts listening for the first time or resumes after a [_pause].
+  void _ensureListening();
+
+  /// Matches events with requests.
+  ///
+  /// Called after receiving an event or when the event source closes.
+  /// May be called by requests which have stalled after the event queue

[nweiz, 2015/08/21 19:43:57]

"stalled" is confusing for me here—it's not a term that's used elsewhere, and it
makes me think of timeouts which I don't think is an accurate understanding. Can
you clarify?

[Lasse Reichstein Nielsen, 2015/08/24 08:31:34]

Done.

+  /// is empty or when the event source is done.
+  void _checkQueues() {
+    while (_requestQueue.isNotEmpty) {
+      if (_requestQueue.first.addEvents(_eventQueue, _isDone)) {
+        _requestQueue.removeFirst();
+      } else {
+        return;
+      }
+    }
+    if (!_isDone) {

[nweiz, 2015/08/21 19:43:57]

Nit: I find it tends to look better when block-level statements are separated by
a newline. Also applies to a few methods below.

[Lasse Reichstein Nielsen, 2015/08/24 08:31:34]

I've added some extra lines (not all between block statements, some include a
line before or after that belongs with the loop/if block, and some places I just
didn't think looked better with empty lines (small functions tend to be look
broken up by the extra lines).

+      _pause();
+    }
+  }
+
+  /// Requests that the event source pauses events.
+  ///
+  /// The event source is restarted by the next call to [_ensureListening].
+  void _pause();
+
+  /// Extracts a stream from the event source and makes this stream queue
+  /// unusable.
+  ///
+  /// Can only be used by the very last request.
+  /// Only used by [rest].
+  Stream _extractStream();
+}
+
+
+/// The default implementation of [StreamQueue].
+///
+/// This queue gets its events from a stream which is listened
+/// to when a request needs events.
+class _StreamQueue<T> extends StreamQueue<T> {
+

[nweiz, 2015/08/21 19:43:57]

Nit: extra newline.

[Lasse Reichstein Nielsen, 2015/08/24 08:31:33]

Done.

+  /// Source of events.
+  final Stream _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;
+
+  _StreamQueue(this._sourceStream) : super._();
+
+  Future _cancel() {
+    if (_isDone) return null;
+    if (_subscription == null) _subscription = _sourceStream.listen(null);
+    var future = _subscription.cancel();
+    _onDone();
+    return future;
+  }
+
+  void _onData(T data) {

[nweiz, 2015/08/21 19:43:57]

For what it's worth, these methods are so small I'd just write them inline in
[_ensureListening].

[Lasse Reichstein Nielsen, 2015/08/24 08:31:33]

Done.

+    _addResult(new Result.value(data));
+  }
+
+  void _onError(error, StackTrace stack) {
+    _addResult(new Result.error(error, stack));
+  }
+
+  void _onDone() {
+    _subscription = null;
+    super._onDone();
+  }
+
   void _ensureListening() {
     assert(!_isDone);
     if (_subscription == null) {
       _subscription =
           _sourceStream.listen(_onData, onError: _onError, onDone: _onDone);
     } else {
       _subscription.resume();
     }
   }
 
-  /// Removes all requests and closes them.
-  ///
-  /// Used when the source stream is done.
-  /// After this, no further requests will be added to the queue,
-  /// requests are immediately served entirely by events already in the event
-  /// queue, if any.
-  void _closeAllRequests() {
-    assert(_isDone);
-    while (_requestQueue.isNotEmpty) {
-      var request = _requestQueue.removeFirst();
-      if (!request.addEvents(_eventQueue)) {
-        request.close(_eventQueue);
-      }
-    }
+  void _pause() {
+    _subscription.pause();
   }
 
-  /// Matches events with requests.
-  ///
-  /// Called after receiving an event.
-  void _checkQueues() {
-    while (_requestQueue.isNotEmpty) {
-      if (_requestQueue.first.addEvents(_eventQueue)) {
-        _requestQueue.removeFirst();
-      } else {
-        return;
-      }
+  Stream<T> _extractStream() {
+    assert(_isClosed);
+    if (_isDone) {
+      return new Stream<T>.empty();
     }
-    if (!_isDone) {
-      _subscription.pause();
+    if (_subscription == null) {
+      return _sourceStream;

[nweiz, 2015/08/21 19:43:57]

Nit: consider making this a single-line if.

[Lasse Reichstein Nielsen, 2015/08/24 08:31:34]

I prefer to not hide significant control flow (return/throw) at the end of an if
(if it's an early bailout, then it's ok).

     }
-  }
-
-  /// Extracts the subscription and makes this stream queue unusable.
-  ///
-  /// Can only be used by the very last request.
-  StreamSubscription _dispose() {
-    assert(_isClosed);
     var subscription = _subscription;
     _subscription = null;
     _isDone = true;
-    return subscription;
+    if (!subscription.isPaused) subscription.pause();

[nweiz, 2015/08/21 19:43:57]

Why pause and re-resume it?

[Lasse Reichstein Nielsen, 2015/08/24 08:31:34]

We need to resume it if it was paused, because the SubscriptionStream won't
resume the subscription.

I want to avoid resuming first because that would cause a "resume the pause",
and pause isn't necessarily free if you have to start/stop low-level IO access.
So, I wanted to resume my pause after the SubscriptionStream was created, and
not before.
This was a simple way to do that - another alternative was to remember whether
it was paused in a bool variable, and then only resume afterwards if necessary.
That's probably more readable, so let's do that.

+    var result = new SubscriptionStream<T>(subscription);
+    subscription.resume();
+    return result;
   }
 }
 
+
+

[nweiz, 2015/08/21 19:43:57]

Nit: three newlines here seems excessive. It's also inconsistent with two
newlines above.

[Lasse Reichstein Nielsen, 2015/08/24 08:31:33]

True, should only be two.

 /// Request object that receives events when they arrive, until fulfilled.
 ///
 /// Each request that cannot be fulfilled immediately is represented by
 /// an `_EventRequest` object in the request queue.
 ///
 /// Events from the source stream are sent to the first request in the
 /// queue until it reports itself as [isComplete].
 ///
 /// When the first request in the queue `isComplete`, either when becoming
 /// the first request or after receiving an event, its [close] methods is
@@ skipping 12 matching lines @@
   /// more events.
   /// The call may keep events in the queue until the requeust is complete,
   /// or it may remove them immediately.
   ///
   /// If the method returns true, the request is considered fulfilled, and
   /// will never be called again.
   ///
   /// This method is called when a request reaches the front of the request
   /// queue, and if it returns `false`, it's called again every time a new event
   /// becomes available, or when the stream closes.
-  bool addEvents(Queue<Result> events);
-
-  /// Complete the request.
-  ///
-  /// This is called when the source stream is done before the request
-  /// had a chance to receive all its events. That is, after a call
-  /// to [addEvents] has returned `false`.
-  /// If there are any unused events available, they are in the [events] queue.
-  /// No further events will become available.
-  ///
-  /// The queue should only remove events from the front of the event queue,
-  /// e.g., using [removeFirst].
-  ///
-  /// If the request kept events in the queue after an [addEvents] call,
-  /// this is the last chance to use them.
-  void close(Queue<Result> events);
+  /// If the function returns `false` when the stream has already closed

[nweiz, 2015/08/21 19:43:57]

"the function" -> "this method" or "it"

What are circumstances where we expect this to return `false` when [isDone] is
`true`? If we don't actually expect that to happen, it would be simpler to just
disallow it, and limit the number of circumstances in which a request will call
private methods on the queue.

[Lasse Reichstein Nielsen, 2015/08/23 11:30:00]

That's exactly for "lookahead". The look-ahead may still wait for the user to
decide on rollback/fast-forward, and the user won't know what to do until after
seeing the end of the stream (like being looking for a "foo" event, only at the
end can you see that there were none).

[nweiz, 2015/08/25 00:38:21]

SGTM

+  /// ([isDone] is true), then the request must call [StreamQueue._checkQueues]
+  /// itself when it's ready to continue.
+  bool addEvents(Queue<Result> events, bool isDone);
 }
 
 /// Request for a [StreamQueue.next] call.
 ///
 /// Completes the returned future when receiving the first event,
 /// and is then complete.
 class _NextRequest<T> implements _EventRequest {
   /// Completer for the future returned by [StreamQueue.next].
   final Completer _completer;
 
   _NextRequest() : _completer = new Completer<T>();
 
   Future<T> get future => _completer.future;
 
-  bool addEvents(Queue<Result> events) {
-    if (events.isEmpty) return false;
-    events.removeFirst().complete(_completer);
-    return true;
-  }
-
-  void close(Queue<Result> events) {
-    var errorFuture =
-        new Future.sync(() => throw new StateError("No elements"));
-    _completer.complete(errorFuture);
+  bool addEvents(Queue<Result> events, bool isDone) {
+    if (events.isNotEmpty) {
+      events.removeFirst().complete(_completer);
+      return true;
+    }
+    if (isDone) {
+      var errorFuture =
+          new Future.sync(() => throw new StateError("No elements"));
+      _completer.complete(errorFuture);
+      return true;
+    }
+    return false;
   }
 }
 
 /// Request for a [StreamQueue.skip] call.
 class _SkipRequest implements _EventRequest {
   /// Completer for the future returned by the skip call.
   final Completer _completer = new Completer<int>();
 
   /// Number of remaining events to skip.
   ///
   /// The request [isComplete] when the values reaches zero.
   ///
   /// Decremented when an event is seen.
   /// Set to zero when an error is seen since errors abort the skip request.
   int _eventsToSkip;
 
   _SkipRequest(this._eventsToSkip);
 
   /// The future completed when the correct number of events have been skipped.
   Future get future => _completer.future;
 
-  bool addEvents(Queue<Result> events) {
+  bool addEvents(Queue<Result> events, bool isDone) {
     while (_eventsToSkip > 0) {
-      if (events.isEmpty) return false;
+      if (events.isEmpty) {
+        if (isDone) break;
+        return false;
+      }
       _eventsToSkip--;
       var event = events.removeFirst();
       if (event.isError) {
         event.complete(_completer);
         return true;
       }
     }
-    _completer.complete(0);
+    _completer.complete(_eventsToSkip);
     return true;
   }
-
-  void close(Queue<Result> events) {
-    _completer.complete(_eventsToSkip);
-  }
 }
 
 /// Request for a [StreamQueue.take] call.
 class _TakeRequest<T> implements _EventRequest {
   /// Completer for the future returned by the take call.
   final Completer _completer;
 
   /// List collecting events until enough have been seen.
   final List _list = <T>[];
 
   /// Number of events to capture.
   ///
   /// The request [isComplete] when the length of [_list] reaches
   /// this value.
   final int _eventsToTake;
 
   _TakeRequest(this._eventsToTake) : _completer = new Completer<List<T>>();
 
   /// The future completed when the correct number of events have been captured.
   Future get future => _completer.future;
 
-  bool addEvents(Queue<Result> events) {
+  bool addEvents(Queue<Result> events, bool isDone) {
     while (_list.length < _eventsToTake) {
-      if (events.isEmpty) return false;
+      if (events.isEmpty) {
+        if (isDone) break;
+        return false;
+      }
       var result = events.removeFirst();
       if (result.isError) {
         result.complete(_completer);
         return true;
       }
       _list.add(result.asValue.value);
     }
     _completer.complete(_list);
     return true;
   }
-
-  void close(Queue<Result> events) {
-    _completer.complete(_list);
-  }
 }
 
 /// Request for a [StreamQueue.cancel] call.
 ///
 /// The request needs no events, it just waits in the request queue
 /// until all previous events are fulfilled, then it cancels the stream queue
 /// source subscription.
 class _CancelRequest implements _EventRequest {
   /// Completer for the future returned by the `cancel` call.
   final Completer _completer = new Completer();
 
   /// The [StreamQueue] object that has this request queued.
   ///
   /// When the event is completed, it needs to cancel the active subscription
   /// of the `StreamQueue` object, if any.
   final StreamQueue _streamQueue;
 
   _CancelRequest(this._streamQueue);
 
   /// The future completed when the cancel request is completed.
   Future get future => _completer.future;
 
-  bool addEvents(Queue<Result> events) {
-    _shutdown();
+  bool addEvents(Queue<Result> events, bool isDone) {
+    _completer.complete(_streamQueue._cancel());
     return true;
   }
-
-  void close(_) {
-    _shutdown();
-  }
-
-  void _shutdown() {
-    if (_streamQueue._subscription == null) {
-      _completer.complete();
-    } else {
-      _completer.complete(_streamQueue._dispose().cancel());
-    }
-  }
 }
 
 /// Request for a [StreamQueue.rest] call.
 ///
 /// The request is always complete, it just waits in the request queue
 /// until all previous events are fulfilled, then it takes over the
 /// stream events subscription and creates a stream from it.
 class _RestRequest<T> implements _EventRequest {
   /// Completer for the stream returned by the `rest` call.
   final StreamCompleter _completer = new StreamCompleter<T>();
 
   /// The [StreamQueue] object that has this request queued.
   ///
   /// When the event is completed, it needs to cancel the active subscription
   /// of the `StreamQueue` object, if any.
   final StreamQueue _streamQueue;
 
   _RestRequest(this._streamQueue);
 
   /// The stream which will contain the remaining events of [_streamQueue].
   Stream<T> get stream => _completer.stream;
 
-  bool addEvents(Queue<Result> events) {
-    _completeStream(events);
-    return true;
-  }
-
-  void close(Queue<Result> events) {
-    _completeStream(events);
-  }
-
-  void _completeStream(Queue<Result> events) {
+  bool addEvents(Queue<Result> events, bool isDone) {
     if (events.isEmpty) {
       if (_streamQueue._isDone) {
         _completer.setEmpty();
       } else {
-        _completer.setSourceStream(_getRestStream());
+        _completer.setSourceStream(_streamQueue._extractStream());
       }
     } else {
       // There are prefetched events which needs to be added before the
       // remaining stream.
       var controller = new StreamController<T>();
       for (var event in events) {
         event.addTo(controller);
       }
-      controller.addStream(_getRestStream(), cancelOnError: false)
+      controller.addStream(_streamQueue._extractStream(), cancelOnError: false)
                 .whenComplete(controller.close);
       _completer.setSourceStream(controller.stream);
     }
-  }
-
-  /// Create a stream from the rest of [_streamQueue]'s subscription.
-  Stream _getRestStream() {
-    if (_streamQueue._isDone) {
-      var controller = new StreamController<T>()..close();
-      return controller.stream;
-      // TODO(lrn). Use the following when 1.11 is released.
-      // return new Stream<T>.empty();
-    }
-    if (_streamQueue._subscription == null) {
-      return _streamQueue._sourceStream;
-    }
-    var subscription = _streamQueue._dispose();
-    subscription.resume();
-    return new SubscriptionStream<T>(subscription);
+    return true;
   }
 }
 
 /// Request for a [StreamQueue.hasNext] call.
 ///
 /// Completes the [future] with `true` if it sees any event,
 /// but doesn't consume the event.
 /// If the request is closed without seeing an event, then
 /// the [future] is completed with `false`.
 class _HasNextRequest<T> implements _EventRequest {
   final Completer _completer = new Completer<bool>();
 
   Future<bool> get future => _completer.future;
 
-  bool addEvents(Queue<Result> events) {
+  bool addEvents(Queue<Result> events, bool isDone) {
     if (events.isNotEmpty) {
       _completer.complete(true);
       return true;
     }
+    if (isDone) {
+      _completer.complete(false);
+      return true;
+    }
     return false;
   }
-
-  void close(_) {
-    _completer.complete(false);
-  }
 }