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

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..e8242ce8dc3b28a4338ca05b0f0db75bd247bf58 100644
--- a/lib/src/stream_queue.dart
+++ b/lib/src/stream_queue.dart
@@ -60,7 +60,7 @@ import "../result.dart";
 ///
 /// 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.
@@ -77,16 +77,7 @@ class StreamQueue<T> {
   // 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.
@@ -103,8 +94,9 @@ class StreamQueue<T> {
   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.
   ///
@@ -115,6 +107,8 @@ class StreamQueue<T> {
   ///
   /// 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();
@@ -216,13 +210,13 @@ class StreamQueue<T> {
     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.
   ///
@@ -242,13 +236,13 @@ class StreamQueue<T> {
       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
@@ -258,98 +252,136 @@ class StreamQueue<T> {
   }
 
   // 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].
-  ///
-  /// Resumes subscription on [_sourceStream], or creates it if necessary.
-  void _ensureListening() {
-    assert(!_isDone);
-    if (_subscription == null) {
-      _subscription =
-          _sourceStream.listen(_onData, onError: _onError, onDone: _onDone);
-    } else {
-      _subscription.resume();
-    }
-  }
-
-  /// Removes all requests and closes them.
+  /// Ensures that we are listening on events from the event source.
   ///
-  /// 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);
-      }
-    }
-  }
+  /// Starts listening for the first time or resumes after a [_pause].
+  void _ensureListening();
 
   /// Matches events with requests.
   ///
-  /// Called after receiving an event.
+  /// 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)) {
+      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).

-      _subscription.pause();
+      _pause();
     }
   }
 
-  /// Extracts the subscription and makes this stream queue unusable.
+  /// 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.
-  StreamSubscription _dispose() {
+  /// 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();
+    }
+  }
+
+  void _pause() {
+    _subscription.pause();
+  }
+
+  Stream<T> _extractStream() {
     assert(_isClosed);
+    if (_isDone) {
+      return new Stream<T>.empty();
+    }
+    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).

+    }
     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
@@ -382,22 +414,10 @@ abstract class _EventRequest {
   /// 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.
@@ -412,16 +432,18 @@ class _NextRequest<T> implements _EventRequest {
 
   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;
   }
 }
 
@@ -443,9 +465,12 @@ class _SkipRequest implements _EventRequest {
   /// 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) {
@@ -453,12 +478,8 @@ class _SkipRequest implements _EventRequest {
         return true;
       }
     }
-    _completer.complete(0);
-    return true;
-  }
-
-  void close(Queue<Result> events) {
     _completer.complete(_eventsToSkip);
+    return true;
   }
 }
 
@@ -481,9 +502,12 @@ class _TakeRequest<T> implements _EventRequest {
   /// 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);
@@ -494,10 +518,6 @@ class _TakeRequest<T> implements _EventRequest {
     _completer.complete(_list);
     return true;
   }
-
-  void close(Queue<Result> events) {
-    _completer.complete(_list);
-  }
 }
 
 /// Request for a [StreamQueue.cancel] call.
@@ -520,22 +540,10 @@ class _CancelRequest implements _EventRequest {
   /// 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.
@@ -558,21 +566,12 @@ class _RestRequest<T> implements _EventRequest {
   /// 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
@@ -581,26 +580,11 @@ class _RestRequest<T> implements _EventRequest {
       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;
   }
 }
 
@@ -615,15 +599,15 @@ class _HasNextRequest<T> implements _EventRequest {
 
   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);
-  }
 }