Add new features to package:async.

lib/src/stream_events.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.streams.stream_events;
+
+import 'dart:async';
+import 'dart:collection';
+
+import "subscription_stream.dart";
+import "stream_completer.dart";
+
+/// An asynchronous pull-based interface for accessing stream events.
+///
+/// Wraps a stream and makes individual events available on request.
+///
+/// The individual requests are served in the order they are requested.

[nweiz, 2015/06/12 01:24:25]

Explain how this is different from a [StreamIterator]. Also, document that the
underlying stream is paused as long as there are no pending requests.

[Lasse Reichstein Nielsen, 2015/06/17 11:08:29]

Done.

+///
+/// Example:
+///
+///     var events = new StreamEvents<String>(someStreamOfLines);
+///     var first = await events.next;
+///     while (first.startsWith('#')) {

[Søren Gjesse, 2015/06/11 07:57:06]

Maybe add an isDone check in this sample.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:22]

Hmm. Not good for the flow, but I guess I can try something.

+///       // Skip comments.
+///       first = await events.next;
+///     }
+///
+///     if (first.startsWith(MAGIC_MARKER)) {
+///       var headerCount =
+///           first.parseInt(first.substring(MAGIC_MARKER.length + 1));
+///       handleMessage(headers: await events.take(headerCount),
+///                     body: events.rest);
+///       return;
+///     }
+///     // Error handling.
+///
+class StreamEvents<T> {

[nweiz, 2015/06/12 01:24:26]

Plural class names always seem weird to me. What do you think about calling this
something like "PullStream" instead?

[Lasse Reichstein Nielsen, 2015/06/12 13:04:22]

It's not a stream, so "PullStream" won't work.
There are other constraints, but something that mentions the events of a stream,
and doesn't end in a nounified verb (no "Puller", "Manager", "Handler", etc.).
I haven't found a good alternative.

I also don't have a problem with plural names :)

The usage of this will be:

  var events = new StreamEvents(someStream);
  var first = await events.next;

I think it works.

[nweiz, 2015/06/16 01:05:23]

Bob suggested "StreamPump", using the water analogy: where a stream flows on its
own accord, you explicitly pull water out of a pump one unit at a time.

[Lasse Reichstein Nielsen, 2015/06/16 13:05:45]

I'm not a great fan of analogies in nameing - it only makes sense if you think
the same way as the person doing the naming. Something more objective would be
better. 
It introduces a new concept that the user need to understand. What is a "Pump"?
Do we have other pumps? Should we? 

Maybe StreamEventQueue? Except it's not a Queue in the collection sense.

[nweiz, 2015/06/16 22:34:11]

All names in programming are analogies at some level, but I take your point:
using existing vocabulary is better.
That's true, but despite being overloaded "queue" is a well-known term in
concurrent programming that fits this concept pretty well. I'm down with it,
although I'd call it "StreamQueue" to be more concise.

[Lasse Reichstein Nielsen, 2015/06/17 11:08:29]

I'm trying out StreamQueue now. It is slightly irksome because it sounds like a
queue of streams. StreamEventQueue would sound like a queue of stream events
(which is closer to the truth) and "event queue" is a concept on its own which
helps distance it from the collection Queue.

Let's see if I can get used to StreamQueue.

+  /// In the initial state, the stream has not been listened to yet.

[nweiz, 2015/06/12 01:24:25]

Nit: The first paragraph of a doc comment should be a one-sentence description,
per https://www.dartlang.org/articles/doc-comment-guidelines/.

+  /// It will be listened to when the first event is requested.
+  /// The `stateData` field holds the stream and the request queue is empty.
+  static const int _INITIAL        =  0;

[nweiz, 2015/06/12 01:24:25]

Nit: since these fields aren't right up next to one another, I don't think
there's much value in adding a bunch of extra whitespace to get them aligned.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:22]

In my editor it still works very well to show the layout.

Comments are muted in color so the structure shows as a clear grid of yellow
boilerplate, white names and purple values, all nicely aligned.

+
+  /// Listening on the stream.
+  /// If the request queue is empty and the subscription isn't done,
+  /// the subscription is paused.
+  /// The `stateData` field holds the subscription.
+  static const int _LISTENING      =  1;
+
+  /// The stream has completed.
+  /// The `stateData` field is `null` and the request queue is empty.
+  static const int _DONE           =  2;
+

[Søren Gjesse, 2015/06/11 07:57:06]

Maybe put an additional comment that 4 is reserved bu _ClOSE_REST below.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:21]

Will do.

+  /// Flag set when [close] is called.
+  /// The `StreamEvents` is closed and no further events can be requested.
+  /// The last request in the queue, if any,
+  /// is completed and is witing to clean up.

[nweiz, 2015/06/12 01:24:24]

"witing" -> "waiting"

[Lasse Reichstein Nielsen, 2015/06/12 13:04:21]

Done.

+  static const int _CLOSED         =  8;
+
+  /// Flag set when [rest] is called.
+  /// Only used for error reporting, otherwise equivalent to [_CLOSED].
+  static const int _CLOSED_REST    = 12;
+
+  /// Current state.
+  ///
+  /// Use getters below to check if the state is [_isListening] or [_isDone],
+  /// and whether the stream events object [_isClosed].
+  int _state = _INITIAL;
+
+  /// Value depending on state. Use getters below to get the value and assert
+  /// the expected state.
+  var _stateData;
+
+  /// Queue of pending requests while state is [_LISTENING].
+  /// Access through methods below to ensure consistency.
+  Queue<_EventRequest> _requestQueue = new Queue();

[nweiz, 2015/06/12 01:24:25]

Make this final.

Personally, I prefer to omit the type annotation for private variables with
initializers, since it's clear from the value what the type is. I'd write this:

    final _requestQueue = new Queue<_EventRequest>();

[Lasse Reichstein Nielsen, 2015/06/12 13:04:21]

Made final.
I prefer type annotations on all fields.

+
+  StreamEvents(Stream source) : _stateData = source;
+
+  bool get _isListening => (_state & _LISTENING) != 0;
+  bool get _isClosed    => (_state & _CLOSED)    != 0;
+  bool get _isDone      => (_state & _DONE)      != 0;
+
+  /// Whether the underlying stream is spent.

[nweiz, 2015/06/12 01:24:24]

"spent" -> "done"

Just to use consistent terminology for the same thing.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:22]

Done.

+  ///
+  /// This may return true before all events have been delivered.
+  /// Requesting a new event when [isDone] returns true,
+  /// for example using [next], will always fail.
+  bool get isDone       => _isDone;

[nweiz, 2015/06/12 01:24:25]

What's the use case for exposing this to the user? It seems like it violates the
encapsulation of the inner stream.

Also, consider adding a [hasNext] getter that (asynchronously) returns whether a
[next] can be called without getting a [StateError]. This will make it easier
for users to safely iterate through the stream.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:23]

It might be a little speculative because it doesn't actually guarantee that
there will be a next data/error event - the next event might be the done event. 

Let's remove it - it looks like it allows you to check if .next will work, which
it doesn't.
I don't think that would work with this abstraction.
We don't know if there is a next event until that event arrives. We are not
buffering events, so we need to put the event somewhere when it arrives.

Maybe have a more complex nextOrElse function:
  Future nextOrElse(orElse());
which returns the next event as it arrives, and completes with the value of
orElse() if the streams is done before that. What "next" does is to make that an
error, but we could allow the user to define what happens in the done case, so
they can recognize it.
The default could be returning null or throwing the same state error as "next".

[nweiz, 2015/06/16 01:05:23]

The version of this I wrote for ScheduledTest has [hasNext], and it works well.
It is the case that it has to wait for an actual event to arrive, but that's the
case anyway when you're calling [next], and it's really nice to have a way to do
so while guaranteeing that it won't throw. It also enables the following
pattern:

    while (... && await events.hasNext) {
      values.add(await events.next);
    }
This would also be useful, but I don't think it obviates the need for [hasNext].
Being able to do while loops like the one above is really nice, as is being able
to check that the object is in a valid state before passing it to something else
that actually calls [next].

[Lasse Reichstein Nielsen, 2015/06/16 13:05:45]

It doesn't solve the problem of what to do with the event value when it arrives
if there is no later event waiting for it.
That requires buffering the event (value or error) until the next request is
made. If the next request is "rest", then it needs to push the event back on the
subscription before creating the SubscriptionStream.

I guess I could do a wrapper class for that, wrapping a subscription with a list
of prior events to provide before the real events start. Still complicated, but
I can see the wish for a `hasNext`.

[nweiz, 2015/06/16 22:34:11]

It's unlikely that someone will call [hasNext] followed immediately by [rest]—at
that point why not just call [rest] on its own and check for an early
completion? So you can probably afford to make that case do a bunch of extra
wrapping to keep it simple. For example:

    var controller = new StreamController(sync: true);
    controller.add(_bufferedEvent);
    controller.addStream(new SubscriptionStream(_subscription));
    return controller.stream;

+
+  /// Return the stream subscription while state is listening.

[nweiz, 2015/06/12 01:24:25]

Nit: use present tense (e.g. "Returns" rather than "Return") when describing
methods. It makes it much easier to avoid confusing tense disagreements.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:22]

Done.

+  StreamSubscription get _subscription {
+    assert(_isListening);
+    return _stateData;
+  }
+
+  /// Return the source stream while state is initial.
+  Stream get _sourceStream {
+    assert(!_isListening);
+    assert(!_isDone);
+    return _stateData;
+  }
+
+  // Set the subscription and transition to listening state.
+  void set _subscription(StreamSubscription subscription) {
+    assert(!_isListening);
+    assert(!_isDone);
+    _stateData = subscription;
+    _state |= _LISTENING;
+  }
+
+  void _setDone() {
+    assert(!_isDone);
+    _state = (_state & _CLOSED_REST) | _DONE;
+    _stateData = null;
+  }
+
+  /// Request the next (yet unrequested) event from the stream.

[nweiz, 2015/06/12 01:24:26]

Clarify in this documentation that it's valid to have multiple pending calls to
[next] at once (this is an important difference from [StreamIterator]).

[Lasse Reichstein Nielsen, 2015/06/12 13:04:21]

Done.

+  ///
+  /// When the requested event arrives, the returned future is completed with
+  /// the event. This is independent of whether the event is a data event or
+  /// an error event.

[nweiz, 2015/06/12 01:24:26]

I'd write this as: "If the event is a value event, the returned future completes
with its value. If it's an error, the returned future throws that error."

[Lasse Reichstein Nielsen, 2015/06/12 13:04:22]

Done.

+  ///
+  /// If the stream closed before an event arrives, the future is completed

[nweiz, 2015/06/12 01:24:24]

"closed" -> "closes", "is completed" -> "completes"

[Lasse Reichstein Nielsen, 2015/06/12 13:04:22]

Done.

+  /// with a [StateError].
+  Future<T> get next {
+    if (!_isClosed) {
+      Completer completer = new Completer<T>();
+      _addRequest(new _NextRequest(completer));
+      return completer.future;
+    }
+    throw _failClosed();
+  }
+
+  /// Request a stream of all the remaning events of the source stream.

[nweiz, 2015/06/12 01:24:25]

"Request" -> "Returns"

[Lasse Reichstein Nielsen, 2015/06/12 13:04:21]

Done.

+  ///
+  /// All requested [next], [skip] or [take] operations are completed
+  /// first, and then any remaining events are provided as events of
+  /// the returned stream.
+  ///
+  /// Using `rest` closes the stream events object. After getting the
+  /// `rest` it is no longer allowed to request other events, like

[nweiz, 2015/06/12 01:24:25]

"it is no longer allowed to" -> "the caller may no longer

[Lasse Reichstein Nielsen, 2015/06/12 13:04:22]

Done.

+  /// after calling [close].
+  Stream<T> get rest {
+    if (!_isClosed) {

[nweiz, 2015/06/12 01:24:25]

Nit: Short-circuit if it's closed to save on indentation.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:22]

ACK.
This is not going to be in any inner loops, so one unpredicted call is unlikely
to be a problem.

+      _state |= _CLOSED_REST;
+      if (_isListening) {
+        // We have an active subscription that we want to take over.
+        var delayStream = new StreamCompleter<T>();
+        _addRequest(new _RestRequest<T>(delayStream, this));
+        return delayStream.stream;
+      }
+      assert(_requestQueue.isEmpty);
+      if (isDone) {
+        // TODO(lrn): Add Stream.empty() constructor.
+        return new Stream<T>.fromIterable(const []);
+      }
+      // We have never listened the source stream, so just return that directly.

[nweiz, 2015/06/12 01:24:26]

"listened to"

[Lasse Reichstein Nielsen, 2015/06/12 13:04:22]

Done.

+      Stream result = _sourceStream;
+      _setDone();
+      return result;
+    }
+    throw _failClosed();
+  }
+
+  /// Requests to skip the next [count] *data* events.

[nweiz, 2015/06/12 01:24:24]

"Requests to skip" -> "Skips"

+  ///
+  /// The [count] value must be non-negative.
+  ///
+  /// When successful, this is equivalent to using [take]
+  /// and ignoring the result.
+  ///
+  /// If an error occurs before `count` data events has

[nweiz, 2015/06/12 01:24:25]

"has" -> "have"

+  /// been skipped, the returned future completes with
+  /// that error instead.
+  ///
+  /// If the stream closes before `count` data events,
+  /// the remaining unskipped event count is returned.
+  /// If the returned future completes with the integer `0`,
+  /// then all events were succssfully skipped. If the value
+  /// is greater than zero then the stream ended early.
+  Future<int> skip(int count) {
+    if (count < 0) throw new RangeError.range(count, 0, null, "count");
+    if (!_isClosed) {
+      Completer completer = new Completer<int>();
+      _addRequest(new _SkipRequest(completer, count));
+      return completer.future;
+    }
+    throw _failClosed();
+  }
+
+  /// Requests the next [count] data events as a list.
+  ///
+  /// The [count] value must be non-negative.

[nweiz, 2015/06/12 01:24:25]

"The [count] value" -> "[count]"

[Lasse Reichstein Nielsen, 2015/06/12 13:04:21]

-> "The [count]".
Otherwise it would start the sentence with a lower-case character. 
That would be bad!

+  ///
+  /// Equivalent to calling [next] `count` times and
+  /// storing the data values in a list.
+  ///
+  /// If an error occurs before `count` data events has
+  /// been collected, the returned future completes with
+  /// that error instead.
+  ///
+  /// If the stream closes before `count` data events,
+  /// the returned future completes with the list
+  /// of data collected so far. That is, the returned
+  /// list may have fewer than [count] elements.
+  Future<List<T>> take(int count) {
+    if (count < 0) throw new RangeError.range(count, 0, null, "count");
+    if (!_isClosed) {
+      Completer completer = new Completer<List<T>>();
+      _addRequest(new _TakeRequest(completer, count));
+      return completer.future;
+    }
+    throw _failClosed();
+  }
+
+  /// Release the underlying stream subscription.

[nweiz, 2015/06/12 01:24:26]

"Release" -> "Cancels"

[Lasse Reichstein Nielsen, 2015/06/12 13:04:22]

Done.

+  ///
+  /// The close operation waits until all previously requested
+  /// events have been processed, then it cancels the subscription
+  /// providing the events.
+  ///
+  /// The returned future completes with the result of calling
+  /// `cancel`.
+  ///
+  /// After calling `close`, no further events can be requested.
+  /// None of [next], [rest], [skip], [take] or [close] may be
+  /// called again.
+  Future close() {

[nweiz, 2015/06/12 01:24:24]

Consider calling this [cancel], to match [StreamSubscription] and
[StreamIterator].

[Lasse Reichstein Nielsen, 2015/06/12 13:04:22]

Good idea!

+    if (!_isClosed) {
+      _state |= _CLOSED;
+      if (!_isListening) {
+        assert(_requestQueue.isEmpty);

[nweiz, 2015/06/12 01:24:24]

Add a comment explaining why the request queue will be empty here.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:22]

Done.

+        if (!_isDone) _setDone();
+        return new Future.value();
+      }
+      Completer completer = new Completer();
+      _addRequest(new _CloseRequest(completer, this));
+      return completer.future;
+    }
+    throw _failClosed();

[nweiz, 2015/06/12 01:24:26]

Everything in the core libraries that's closable allows [close] to be called
multiple times; it seems like this should be similar.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:22]

True.
One reason why I didn't allow that was that calling "cancel" after calling
"rest" would not cancel anything.

That's not necessarily a problem - we have the state and knows that 'rest' or
'cancel' has been called already, so we can just do nothing.

I'll see if I can change it so (now) cancel can be called more than once.
Heck, we can even allow you to call rest more than once (getting an empty stream
each time after the first - but that might not be that iteresting.

+  }
+
+  /// Reused error message.

[nweiz, 2015/06/12 01:24:24]

Expand on this.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:21]

Done.

+  Error _failClosed() {
+    String cause =
+        ((_state & _CLOSED_REST) == _CLOSED_REST) ? "rest" : "close";
+    return new StateError("Already closed by a call to $cause");
+  }
+
+  /// Called when requesting an event when the requeust queue is empty.
+  /// The underlying subscription is paused in that case, except the very
+  /// first time when the subscription needs to be created.
+  void _listen() {

[nweiz, 2015/06/12 01:24:24]

Methods like this and [_pause] that are only called from one place and have a
lot of description to explain the context seem like they might be clearer if
they were just inlined.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:23]

Done.

+    if (_isListening) {
+      _subscription.resume();
+    } else if (!_isDone) {
+      _subscription =
+          _sourceStream.listen(_onData, onError: _onError, onDone: _onDone);
+    }
+  }
+
+  /// Pauses the underlying subscription.
+  /// Called when the request queue is empty and the subscription isn't closed.
+  void _pause() {
+    assert(_isListening);
+    _subscription.pause();
+  }
+
+  // Callbacks receiving the events of the source stream.

[nweiz, 2015/06/12 01:24:25]

Nit: If this is referring to a group of methods, add a blank line after it so
that it doesn't look like it's just attached to [_onData].

[Lasse Reichstein Nielsen, 2015/06/12 13:04:21]

Done.

+  void _onData(T data) {
+    assert(_requestQueue.isNotEmpty);
+    _EventRequest action = _nextAction;
+    action.add(data);
+    _checkCompleted();
+  }
+
+  void _onError(error, StackTrace stack) {
+    assert(_requestQueue.isNotEmpty);
+    _EventRequest action = _nextAction;
+    action.addError(error, stack);
+    _checkCompleted();
+  }
+
+  void _onDone() {
+    _setDone();
+    _closeAllRequests();
+  }
+
+  // Request queue management.
+
+  /// Get the next action in the queue, but don't remove it.
+  _EventRequest get _nextAction {
+    assert(_requestQueue.isNotEmpty);

[nweiz, 2015/06/12 01:24:24]

This assertion seems redundant, since the line below will throw if the queue is
empty anyway.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:22]

True.

+    return _requestQueue.first;
+  }
+
+  /// Add a new request to the queue.
+  void _addRequest(_EventRequest action) {
+    if (_isDone) {
+      action.close();
+      return;
+    }
+    if (_requestQueue.isEmpty) {
+      if (action.isComplete) {

[nweiz, 2015/06/12 01:24:24]

Explain under what circumstances an action that was just added could already be
complete.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:22]

Done.

+        // Skip listening and complete this immediately.
+        action.close();
+        return;
+      }
+      _listen();
+    }
+    _requestQueue.add(action);
+  }
+
+  /// Remove all requests and call their `done` event.
+  ///
+  /// Used when the source stream dries up.
+  void _closeAllRequests() {
+    assert(_isDone);
+    while (_requestQueue.isNotEmpty) {
+      _requestQueue.removeFirst().close();
+    }
+  }
+
+  /// Check whether the next actions in the queue are complete.
+  ///
+  /// If so, remove them and call their `complete` method.
+  void _checkCompleted() {
+    // Close-actions are executed immediately when they become the
+    // next (and last) event in the queue.
+    // When _isClosed and the queue is not empty, the last element
+    // of the queue is the close action.
+    while (_requestQueue.isNotEmpty) {
+      if (!_requestQueue.first.isComplete) {
+        return;
+      }
+      _requestQueue.removeFirst().close();
+    }
+    assert(_requestQueue.isEmpty);
+    if (!_isDone) _pause();
+  }
+
+  /// Extracts the subscription and makes the events object unusable.
+  ///
+  /// Can only be used by the very last request.
+  StreamSubscription _dispose() {
+    assert(_isClosed);
+    assert(_isListening);
+    assert(_requestQueue.isEmpty);
+    StreamSubscription subscription = _subscription;
+    _setDone();
+    return subscription;
+  }
+}
+

[nweiz, 2015/06/12 01:24:25]

Nit: extra newline.

+
+/// Action to take when a requested event arrives.
+abstract class _EventRequest implements EventSink {

[nweiz, 2015/06/12 01:24:25]

It's kind of weird that the documentation for these describes them as "actions"
but the class names are all "request". It would be nice to settle on consistent
terminology.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:22]

Agree, I switched to "request" halfway through, and haven't fixed all
references.

+  bool get isComplete;

[Søren Gjesse, 2015/06/11 07:57:06]

Add close method here as well

I think the 'close' method should have a different name, e.g. 'done'.

[nweiz, 2015/06/12 01:24:25]

This should also have [add] and [addError] abstract methods.

Also, document this.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:22]

It's inherited from EventSink.

I picked EventSink as the interface because I consider making a public
_addRequest so you can create custom event collectors. It might not make a
difference because they will have to be wrapped anyway, but it's simple to use
the standard names for now.

+}
+
+/// Action completing a [StreamEvents.next] request.
+class _NextRequest implements _EventRequest {
+  Completer completer;

[nweiz, 2015/06/12 01:24:25]

Even though it doesn't do anything at the language level, I really like making
members of private classes that aren't used by other classes private. It helps
communicate to the reader that this isn't something that will be used
externally.

+  _NextRequest(this.completer);

[nweiz, 2015/06/12 01:24:24]

Rather than passing in a completer, consider making it an instance field and
exposing its future to the caller. For example:

    Future get future => _completer.future;
    final _completer = new Completer();

then from the caller:

    var request = new _NextRequest();
    _addRequest(request);
    return request.future;

This just helps keep the code a little more cleanly layered.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:23]

Good idea. Will do.
The type parameter needs to go on the request class too then, but that's ok.

+
+  void add(data) {
+    completer.complete(data);
+    completer = null;

[nweiz, 2015/06/12 01:24:24]

Does nulling this out buy you anything? It seems like you could just check
[completer.isCompleted] instead of checking against null.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:23]

We have an isCompleted getter? Whoa!
I'm too used to work directly on _Future objects :)

+  }
+
+  void addError(error, [StackTrace stack]) {
+    completer.completeError(error, stack);
+    completer = null;
+  }
+
+  void close() {
+    if (!isComplete) {
+      completer.completeError(new StateError("no elements"));

[nweiz, 2015/06/12 01:24:25]

If you do continue to check against null, null out completer here.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:22]

Technically not necessary since "close" is always the last method called on a
request object - but better for readability.

Anyway, rewritten to not use null as a signal.

+    }
+  }
+
+  bool get isComplete => completer == null;
+}
+
+/// Action completing a [StreamEvents.skip] request.
+class _SkipRequest implements _EventRequest {
+  final Completer completer;
+  int count;

[nweiz, 2015/06/12 01:24:24]

Document this. Also consider calling it something more semantic, like
"eventsToSkip".

[Lasse Reichstein Nielsen, 2015/06/15 15:46:23]

Done.

+  _SkipRequest(this.completer, this.count);
+
+  void add(data) {
+    count--;
+  }
+
+  void addError(error, [StackTrace stack]) {
+    completer.completeError(error, stack);
+    count = 0;
+  }
+
+  void close() {
+    if (!completer.isCompleted) {
+      completer.complete(count);
+      count = 0;
+    }
+  }
+
+  bool get isComplete {
+    return count == 0;
+  }
+}
+
+/// Action completing a [StreamEvents.take] request.
+class _TakeRequest<T> implements _EventRequest {
+  final Completer completer;
+  final List list = <T>[];
+  int count;
+  _TakeRequest(this.completer, this.count);
+
+  void add(data) {
+    list.add(data);
+    count--;
+  }
+
+  void addError(error, [StackTrace stack]) {
+    completer.completeError(error, stack);
+    count = 0;
+  }
+
+  void close() {
+    if (!completer.isCompleted) {
+      completer.complete(list);
+    }
+  }
+
+  bool get isComplete => count == 0;
+}
+
+/// Action completing a [StreamEvents.close] request.
+class _CloseRequest implements _EventRequest {
+  final Completer completer;
+  StreamEvents events;
+
+  _CloseRequest(this.completer, this.events);
+
+  void add(data) {
+    throw new UnsupportedError("event");
+  }
+
+  void addError(error, [StackTrace stack]) {
+    throw new UnsupportedError("event");

[nweiz, 2015/06/12 01:24:24]

Shouldn't this be "error"?

[Lasse Reichstein Nielsen, 2015/06/12 13:04:22]

No, we don't support receiving events, that's the problem.
Maybe it should really just be "assert(false);" because the functions are/should
be unreachable.

+  }
+
+  void close() {
+    if (events._isListening) {
+      completer.complete(events._dispose().cancel());
+    } else {
+      completer.complete();
+    }
+  }
+
+  bool get isComplete => true;
+}
+
+/// Action completing a [StreamEvents.rest] request.
+class _RestRequest<T> implements _EventRequest {
+  final StreamCompleter delayStream;
+  final StreamEvents events;
+  _RestRequest(this.delayStream, this.events);
+
+  void add(data) {
+    throw new UnsupportedError("event");
+  }
+
+  void addError(error, [StackTrace stack]) {
+    throw new UnsupportedError("event");

[nweiz, 2015/06/12 01:24:26]

Ditto.

[Lasse Reichstein Nielsen, 2015/06/15 15:46:23]

Done.

+  }
+
+  void close() {
+    if (events._isListening) {
+      StreamSubscription subscription = events._dispose();
+      delayStream.setSourceStream(new SubscriptionStream<T>(subscription));
+      if (subscription.isPaused) subscription.resume();
+    } else {
+      assert(events._isDone);
+      delayStream.setEmpty();
+    }
+  }
+
+  bool get isComplete => true;
+}