syntax support for Stream programming

[DartBot]

This issue was originally filed by @seaneagan

---

In the same way that programming with Futures could benefit from language support (issue #104), programming with Streams could as well. Streams are essentially async Iterables, so an async version of the "for" statement could work for Streams. It probably makes sense to call it "on" which is already a contextual keyword elsewhere. As an example the implementation of the Iterable-based methods on Stream could look identical to how they are in Iterable except replacing "on" with "for":

async bool any(bool f(E e)) {
  on(var e in this) {
    if(f(e)) return true;
  }
  return false;
}

notes:

* the "on" statement would trigger a "subscribe" to the Stream passing onData, onDone, and onError resulting in a StreamSubscription

• as with await, the enclosing function would implicitly return a Future
• as with await, errors within the construct would throw the unwrapped error, not the AsyncError wrapper
• any "break" or "return" in the "on" statement would trigger an "unsubscribe" on the StreamSubscription
• nested "await"s and "on"s could be delimited by "pause()" and "resume()" on the StreamSubscription, and since that's only a best effort, any events which occur before the nested items complete could be buffered for use in later iterations of the "on" statement.

Initial mailing list discussion:

http://goo.gl/8PqcP

[anders-sandholm]

cc @floitschG.
Added Area-Language, Triaged labels.

[DartBot]

This comment was originally written by [email protected]

---

Thank you for the excellent issue description!

[DartBot]

This comment was originally written by @seaneagan

---

Related idea... async generators (issue #36 is sync generators). Functions which use (await OR on (see above)) AND yield would return Streams.

Examples:

1. An easier way to do StreamTransformers:

Stream everyOtherEvent(Stream stream) {
  int i = 0;
  on(var e in Stream) if(i.isEven) yield e;
}

2. Flexible Stream generation:

/// Think... "Connection error... retrying in <x seconds>"
Stream<Duration> retry(action, Duration initial, {int max}) {
  var duration = initial;
  var tries = 0;
  while(max == null || tries++ <= max) {
    if(action()) return;
    yield await new Future.delayed(duration);
    duration *= 2;
  }
  throw '...';
}

[gbracha]

We are looking at adding such support as part of the async-await proposal.

---

Set owner to @gbracha.
Added Accepted label.

[gbracha]

The Dart spec now includes async* methods and await for loops. Not fully implemented yet, but in progress. I'm closing this bug as "Done" from the language perspective.

---

Added Done label.