Add new features to package:async.

lib/src/stream_completer.dart

Index: lib/src/stream_completer.dart
diff --git a/lib/src/stream_completer.dart b/lib/src/stream_completer.dart
new file mode 100644
index 0000000000000000000000000000000000000000..9d743a183043383c46bdcd144f7663b36c343ae1
--- /dev/null
+++ b/lib/src/stream_completer.dart
@@ -0,0 +1,275 @@
+// Copyright (c) 2015, the Dart project authors.  Please see the AUTHORS file

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

As part of moving a bunch of existing async utilities to this package, I
realized I had a StreamGroup class that takes a bunch of streams and merges
their outputs. Its functionality is a superset of this and could be used to
implement it very straightforwardly. It would probably be a little less
efficient due to more indirection, but it would cut down on the code size a lot.

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

True. And if I was just using a controller this could be much simpler:

class StreamCompleter {
 var _controller;
 bool _listened = false;
 var _stream;
 StreamCompleter() {
   _controller = new StreamController(sync: true,
                                      onListen: _listen);
 }
 Stream get stream => _controller.stream;
 void _listen() {
   _listened = true;
   if (_stream != null) {
     _stream.pipe(_controller);
   }
 }
 void setSourceStream(Stream source) {
  if (_stream != null) throw new StateError("");
    _stream = source;
    if (_listened) {
      if (_controller.hasListener) {
        source.pipe(_controller);
      } else {
        // Has already been listened to and canceled again.
        source.listen(null).cancel().catchError((_){});
      }
    }
  }
}

Maybe I'm trying too hard to avoid overhead? Nah! Impossible!

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

Here's a (rough untested) sketch of what it would look like with StreamGroup:

class StreamCompleter {
  final _group = new StreamGroup();

  Stream get stream => _group.stream;

  Future setSourceStream(Stream source) {
    var future = _group.add(source);
    _group.close();
    return future;
  }
}

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

And, to make it even more obvious that this is a good idea, I actually do want a
StreamGroup instead of a StreamCompleter for implementing hasNext, because then
I can just add a stream of the initial event, and then the remaining stream.

The specialized StreamCompleter can still optimize the one case where you set
the stream before listening, where you can forward the listen directly to the
source stream without an intermediate controller. 

So, please do add StreamGroup!

[nweiz, 2015/06/17 22:58:45]

Oh dang, I thought I published that CL yesterday, but I forgot. Here it is:
https://codereview.chromium.org/1178793006/

+// 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_completer;
+
+import 'dart:async';
+
+/// A [stream] where the contents aren't known at creation time.
+///
+/// It is generally recommended that you never create a `Future<Stream>`
+/// because you can just use directly create a stream that doesn't do anything

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

"use directly" -> "directly"

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

Done.

+/// until it's ready to do so.
+/// This class can be used to create such a stream.
+///
+/// The [stream] is a normal stream that you can listen to immediately,
+/// but until [setSourceStream] is called, the stream won't produce
+/// any events.
+///
+/// The same effect can be achieved by using a [StreamController]
+/// and adding the stream using `addStream` when both
+/// the controller's stream is listened to and the source stream is ready.
+/// This class attempts to shortcut some of the overhead when possible.
+/// For example, if the [stream] is only listened to
+/// after the source stream has been set,
+/// the listen is performed directly on the source stream.
+class StreamCompleter<T> {

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

Maybe add a constructor with an optional onListen callback taking the
StreamCompleter as an argument. That callback can then call setSourceStream when
the stream is listened on.

Use case: The resulting stream is from some async request where the response
completes with something that has/is a stream (e.g. HTTP or other RPC request).

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

Consider adding a constructor so that the user can specify whether this will
return a broadcast stream or not.

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

I think I'd use a completely different class for a broadcast stream,
that would be BroadcastStreamCompleter.
That's necessary because a broadcast stream can have multiple simultaneous
subscriptions, and the stream here doesn't handle that.

I'll change the documentation to say "single-subscription".

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

You are thinking the case where you don't even want to create the stream if it
won't be used.
Makes sense, will consider.

+  final Stream<T> stream = new _PromiseStream<T>();

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

Nit: document this

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

Done.

+

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

Suggestion: add "static Stream fromFuture(Future<Stream> future)". I've written
that function several times across several projects, and it would be nice to
have it available here.

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

Good idea. Done.

+  /// Set a stream as the source of events for the [StreamCompleter].
+  ///
+  /// There is no guarantee that the stream will ever be listened to.

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

It would be nice to go further and fully explain how all operations—listening,
pausing, canceling—on the returned stream affect the source stream. Also,
document what happens if multiple source streams are set.

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

Done.

+  void setSourceStream(Stream<T> stream) {
+    _PromiseStream promiseStream = this.stream;
+    promiseStream._linkStream(stream);
+  }
+
+  /// As setting an empty stream using [setSourceStream].
+  void setEmpty() {

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

Document that this is mutually exclusive with [setSourceStream].

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

Done.

+    // TODO(lrn): Optimize this to not actually create the empty stream.
+    _PromiseStream promiseStream = this.stream;
+    promiseStream._linkStream(new Stream.fromIterable(const []));
+  }
+}
+
+/// Stream that acts as a source stream it is (eventually) linked with.
+///
+/// The linked source stream can be set after a user has started listening on
+/// this stream. No events occur before the source stream is provided.
+///
+/// If a user listens before events are available, the state of the
+/// subscription is maintained, and the subscription is then linked
+/// to the source stream when that becomes available.
+class _PromiseStream<T> extends Stream<T> {

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

Even though it's not public, it would be nice to have this name match up with
"StreamCompleter" somehow. Maybe "_CompleterStream"?

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

True. I renamed the public class at some point, so the private names no longer
make sense.

+  static const int _UNINITIALIZED = 0;
+  static const int _STREAM_SET = 1;
+  static const int _LISTENED = 2;

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

Document the semantics of each of these variables.

Nit: Constants should no longer be all-caps per the style guide.

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

Done
I have rewritten this file quite a lot.
It actually works with lower-case constants when I use them as state names.

+
+  int _state = _UNINITIALIZED;

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

Document that this is a bitfield. At first glance, it looks like it just has a
simple enum value.

Actually, it would probably be clearest if [streamSet] and [listened] were both
boolean fields, and [uninitialized] was defined as "bool get uninitialized =>
!streamSet && !listened".

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

This is no longer used as a bit field. I have constants for all four state
combinations, and use those instead of the individual bits.

+  var _streamOrSubscription;

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

This really seems like it should be two separate fields. I get that they'll
never be set at the same time, but merging them together sacrifices readability
for a performance gain that's probably extremely marginal.

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

I think the way I have (now) encapsulated it should make it readable.
You should never read this field directly, and instead use a _stream or
_subscription getter that tests that you are in a compatible state.

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

It's definitely better, but it's still a lot of extra machinery for a very small
benefit.

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

True. I think I'll try rewriting it without the state.

+
+  void _linkStream(Stream stream) {

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

It would be really nice to have all these private functions documented. Since
this is in a package, more people (including me!) are likely to want to look
through the implementation and figure out what's going on.

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

Done.

+    if (_state == _UNINITIALIZED) {
+      _streamOrSubscription = stream;
+      _state |= _STREAM_SET;
+    } else if (_state == _LISTENED) {
+      _PromiseSubscription promiseSubscription = _streamOrSubscription;
+      promiseSubscription._linkStream(stream);
+      _state |= _STREAM_SET;
+    } else {
+      throw new StateError("Stream already linked.");
+    }
+  }
+
+  StreamSubscription listen(void onData(T event),
+                            {Function onError,
+                             void onDone(),
+                             bool cancelOnError}) {
+    int state = _state;

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

Nit: don't type-annotate local variables. There are a couple more scattered
about that aren't necessary for type coercion.

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

Yeah, about that. 
I try to use more "var", but I generally still prefer to use the type if it's at
all necessary (if the type isn't obvious from either the name or the right-hand
side, or if I call any function on the object that isn't on object, so I can get
a warning if it's not correct).

For the state, it's now unnecessary since I don't do integer operations on them.

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

As Bob likes to say, code is for people to read it; if people can figure out
what's going on and tools can't, the tools should be fixed, not the code itself.

Of course, this could be (and has been) debated endlessly, but luckily we have a
style guide to make the final decision for us. And it's pretty unambiguous:
except in exceptional circumstances, local variables shouldn't be
type-annotated.

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

True. 
It's mainly for myself. I like to see the type of a variable declaration without
having to look at the right-hand side.

Good naming goes a long way towards fixing that - then you can just look a the
variable, not even the declaration.

+    _state = state | _LISTENED;
+    if (state == _UNINITIALIZED) {

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

Why not just check (_state & _STREAM_SET) == 0? Then you don't need the temp
variable state.

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

I still need to read the state before setting it, to check that the stream
hasn't already been listened to. It won't buy anything in practice.

+      _PromiseSubscription subscription =
+          new _PromiseSubscription<T>(true == cancelOnError);

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

"true == cancelOnError" is really hard to read. It would be a lot clearer if you
just did "if (cancelOnError == null) cancelOnError = true" at the top of the
function to match the convention for default values.

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

Now forwards the parameter all the way to the final listen, so I never need to
coerce it to bool.

+      subscription.onData(onData);
+      subscription.onError(onError);
+      subscription.onDone(onDone);
+      _streamOrSubscription = subscription;
+      return subscription;
+    }
+    if (state == _STREAM_SET) {
+      Stream stream = _streamOrSubscription;
+      return stream.listen(onData, onError: onError, onDone: onDone,
+                           cancelOnError: cancelOnError);
+    }
+    print(state);

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

Debug print.

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

Acknowledged.

+    throw new StateError("Stream has already been listened to.");
+  }
+}
+
+/// Subscription for a [_PromiseStream] that is listened to but has no data.

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

"has no data" is a little misleading, since this won't be used if the stream has
been completed but the underlying stream hasn't emitted any data. Maybe "hasn't
yet completed to an actual stream"?

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

I have reworded it. It's a little constrained by wanting to fit the core
description on one line, but I think it works.

+///
+/// Maintains the state of a stream subscription that hasn't received any
+/// events until events are available, then it starts forwarding to another

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

"events are available" has the same issue as above.

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

reworded.

+/// subscription.
+class _PromiseSubscription<T> implements StreamSubscription<T> {
+  // State values
+  // The subscription is in one of three distinct states:
+  // - Initial (remembers whether it's cancelOnError and paused).
+  // - Cancelled (before being linked).
+  // - Linked (before being cancelled).
+  static const int _INIT = 0;
+  static const int _INIT_CANCEL_ON_ERROR = 1;
+  static const int _CANCELLED = 2; // Exclusive.
+  static const int _LINKED = 3;  // Exclusive
+  static const int _PAUSE = 4;  // Used with _INIT or _INIT_CANCEL_ON_ERROR.

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

Similarly to the above, it would be a lot clearer if [pause] and [cancelOnError]
were separate fields. Bitwise boolean logic is really hard to understand, and I
can't imagine the performance difference will matter in practice.

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

I have moved the entire state into a separate object so this class is just
delegating to another stream subscription in all cases, and then when you link a
stream the delegate is replaced.
The other class does not have this complex state.

+
+  /// State represents the status of the subscription until the
+  /// real subscription becomes available.
+  ///
+  /// While `_state` is not `_LINKED` or `_CANCELED`, `_stateData` contains
+  /// a list of length three with the data, error and done handlers that
+  /// have been set.
+  ///
+  /// While `_state` is [_LINKED], [_stateData] contains the real
+  /// stream subscription.
+  ///
+  /// When `_state` is `_CANCELED`, `_stateData` is cleared since the
+  /// event handlers won't be needed anyway.
+  int _state;
+  var _stateData = new List(3);

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

Why use a list with three members instead of three named member variables?

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

+1

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

Space! Saving of!
I started writing it as a real object instead, with onData, onError and onDone
fields. Then I realized that cancelOnError should also be on that object. And I
might as well put the pause state there too, and then I had a completely
different (and better) solution. So, thanks :)

+
+  _PromiseSubscription(bool cancelOnError)
+      : _state = (cancelOnError ? _INIT_CANCEL_ON_ERROR : _INIT);
+
+  bool get _isLinked => _state == _LINKED;

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

If you relay want to use a bit-field why not ave getters and settes for all
bits?

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

It's not always a bit field.
It's really a combo:
_CANCELLED and _LINKED are just states.
The remaining case is a cancel-on-error flag and a pause count.

Anyway, it's gone now :)

+  bool get _isCancelled => _state == _CANCELLED;
+  bool get _isInitial => (_state & (_PAUSE - 1)) <= _INIT_CANCEL_ON_ERROR;
+
+  void _linkStream(Stream stream) {
+    if (_isLinked) {
+      throw new StateError("Already linked to a stream.");
+    }
+    if (_isCancelled) {

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

Shouldn't we listen on and cancel the stream in this case?

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

Probably not.

For one thing, it's unnecessary overhead. Why start doing something potentially
expensive if you *know* you are not going to need it.

Also, calling cancel may give you a future with an error, and we don't have any
place to report that error. You already got a completed future from the original
cancel call.

There might be cases where we do want it, so maybe it should be a flag somewhere
to always subscribe, even if canceled.

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

I agree with Søren that the default should be to subscribe and then immediately
cancel; the completer's pattern of interaction with the stream it's linked to
should have as little dependency on ordering as possible.

If we're going to make it configurable, I think it would be better for the
default to be easier to reason about but slightly less efficient, and allow
users to opt in to the more difficult but potentially faster version.

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

Ok. The only problem with that is that listen+cancel may return a future with an
error that we have to swallow. That won't happen if we don't listen at all.

+      return;
+    }
+    bool cancelOnError = (_state & _INIT_CANCEL_ON_ERROR) != 0;
+    StreamSubscription subscription =
+        stream.listen(null, cancelOnError: cancelOnError);
+    List handlers = _stateData;

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

Why this 'cast' - documentation?

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

Because I like it. It's not obvious what the type is from either name or
right-hand-side.
And I get warnings if I use it badly below. I like warnings more than hints.

+    subscription.onData(handlers[0]);
+    subscription.onError(handlers[1]);
+    subscription.onDone(handlers[2]);
+    int state = _state;
+    _subscription = subscription;
+    while (state >= _PAUSE) {
+      subscription.pause();
+      state -= _PAUSE;

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

I have no idea what's going on here :-/. If you're going to do complex bitwise
stuff like this, please document *heavily* why it works.

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

Rewritten completely (twice), and there is no fancy stuff left.

+    }
+  }
+
+  List get _handlers {
+    assert(_isInitial);
+    return _stateData;
+  }
+
+  StreamSubscription get _subscription {
+    assert(_isLinked);
+    return _stateData;
+  }
+
+  // Sets state to linked.
+  void set _subscription(StreamSubscription subscription) {

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

Since this is only called once, I think it would be clearer to inline it. Right
now when reading [_linkStream] it looks like it's just assigning to an instance
variable; it's not clear that [_state] gets modified too.

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

Now a function named setListened that does the state transition from _inital to
_listenerOnly;

+    assert(_isInitial);
+    _stateData = subscription;
+    _state = _LINKED;
+  }
+
+  void onData(void handleData(T data)) {
+    if (_isLinked) {
+      _subscription.onData(handleData);
+    } else {
+      assert(_isInitial);
+      _handlers[0] = handleData;
+    }
+  }
+
+  void onError(void handleError(error, StackTrace stackTrace)) {
+    if (_isLinked) {
+      _subscription.onError(handleError);
+    } else {
+      assert(_isInitial);
+      _handlers[1] = handleError;
+    }
+  }
+
+  void onDone(void handleDone()) {
+    if (_isLinked) {
+      _subscription.onDone(handleDone);
+    } else {
+      assert(_isInitial);
+      _handlers[2] = handleDone;
+    }
+  }
+
+  void pause([Future resumeFuture]) {
+    if (_isLinked) {
+      _subscription.pause(resumeFuture);
+    } else if (!_isCancelled) {

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

assert that it is not already paused (or use |).

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

Pauses are cumulative, it can be paused more than once.

+      _state += _PAUSE;
+      if (resumeFuture != null) {
+        resumeFuture.whenComplete(this.resume);
+      }
+    }
+  }
+
+  void resume() {
+    if (_isLinked) {
+      _subscription.resume();
+    } else if (_state >= _PAUSE) {

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

Just check the bit instead of using >=.

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

It can be paused more than once, so the bit will only be set for an odd number
of pauses.

+      _state -= _PAUSE;
+    }
+  }
+
+  Future cancel() {

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

What is the semantics of calling cancel several times? Should it return the same
future?

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

I'm not sure it's specified, but that's what we usually do.
At least it should return equivalent futures - in this case it's pre-completed
futures with a null value.

(I actually have a change for the Future.value constructor so it returns the
same future every time you call it with null as argument - it happens often
enough that it's worth optimizing, and we already have a constant null future in
the code).

+    if (_isLinked) {
+      return _subscription.cancel();
+    } else {
+      _stateData = null;
+      _state = _CANCELLED;
+      return new Future.value();
+    }
+  }
+
+  Future asFuture([futureValue]) {
+    if (_isLinked) {
+      return _subscription.asFuture(futureValue);
+    }
+    Completer completer = new Completer();
+    if (!_isCancelled) {

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

Nit: Short-circuit if it is cancelled to save some indentation.

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

Rewritten.

+      // Asking for a future of a cancelled subscription gives a future
+      // which never completes.
+      _handlers[1] = _cancelBeforeError(completer.completeError);
+      if (futureValue == null) {
+        _handlers[2] = completer.complete;
+      } else {
+        _handlers[2] = () { completer.complete(futureValue); };

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

Nit: Either use "=>" or make this multiple lines.

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

I skipped the special casing of null completely (and uses multiple lines now - I
never user => with a void result).

+      }
+    }
+    return completer.future;
+  }
+
+  bool get isPaused {
+    if (_isLinked) {
+      return _subscription.isPaused;
+    }
+    return _state >= _PAUSE;
+  }
+
+  /// Helper function used by [asFuture].
+  ///
+  /// Returns an error handler which cancels the stream when it receives an
+  /// error.
+  Function _cancelBeforeError(Function handleError) {
+    return (e, s) {

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

Nit: use full words for variables.

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

Done.

+      cancel();
+      if (handleError is _BinaryCallback) {
+        handleError(e, s);
+      } else {
+        handleError(e);
+      }
+    };
+  }
+}
+
+typedef _BinaryCallback(a, b);

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

We don't have this typedef publid in and dart: library?

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

We do: ZoneBinaryCallback in dart:async.
Also not using it any more :)