Add reactive-inspired stream transformers:

lib/src/stream_transformers.dart

+// Copyright (c) 2016, the Dart project authors.  Please see the AUTHORS file

[nweiz, 2016/01/29 22:19:44]

These transformers should all have explicit documentation of their behavior upon
encountering errors.

[Lasse Reichstein Nielsen, 2016/02/01 12:43:20]

True.

+// 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.
+
+import "dart:async";
+import "dart:collection";
+
+/// A group created by [GroupBy].
+///
+/// The stream created by [GroupBy] emits a [Group] for each distinct key
+/// it sees.
+/// This group contains the key itself, along with a stream of the values
+/// associated with that key.
+class Group<K, V> {
+  final K key;

[nweiz, 2016/01/29 22:19:45]

It would be good to have some documentation for these fields, even if it's just
cursory. It makes the dartdoc look a lot better.

[Lasse Reichstein Nielsen, 2016/02/25 16:18:09]

Done.

+  final Stream<V> values;
+  Group(this.key, this.values);
+}
+
+/**
+ * Groups events by a computed key.
+ *
+ * A key is extracted from incoming events, and a stream is created for each
+ * unique key. A value is computed from the event as well, and emitted on the
+ * corresponding stream.
+ *
+ * An error on the source stream, or when calling the `key` or `value`
+ * functions, will emit the error on the returned stream and stop
+ * processing, closing all the individual group streams.
+ *
+ * The returned stream emits a [Group] object for each distinct key seen
+ * by the transformation, and the values associated with the key are output
+ * on the [Group.values] stream.
+ */

[nweiz, 2016/01/29 22:19:45]

Nit: These should be "///" comments.

[Lasse Reichstein Nielsen, 2016/02/25 16:18:10]

Done.

+class GroupBy<S, K, V> implements StreamTransformer<S, Group<K, V>> {

[nweiz, 2016/01/29 22:19:45]

Personally, I'd put each of these into its own file (probably under
lib/src/stream_transformers/). I'd rather have many small, focused file than one
with a bunch of different stuff in it.

[Lasse Reichstein Nielsen, 2016/02/25 16:18:10]

Done.

+  final Function _key;
+  final Function _value;
+
+  /// Groups values returned by [element] by the key returned by [key].

[nweiz, 2016/01/29 22:19:45]

"element" -> "value"

[Lasse Reichstein Nielsen, 2016/02/25 16:18:10]

Done.

+  ///
+  /// If [value] is omitted, it defaults to an identity function, which
+  /// will only work if all the source events are instances of [V].
+  GroupBy(K key(S source), [V value(S source)])

[nweiz, 2016/01/29 22:19:44]

Can we make [value] a named parameter? It makes the API more flexible for the
future, it's similar to [Map.fromIterable], and it's easier to understand
multiple anonymous functions if one of them has a parameter name.

[Lasse Reichstein Nielsen, 2016/02/01 12:43:20]

We could probably make both named parameters.

(Making things "flexible for the future" by making all optional parameters named
is ... not something I favor. It's a fault in the language that you can't have
both kinds of optional parameters on the same function, and making something
that should be positional be named is not picking the best possible API for the
actual job).
Those are good points, though.
I think making both named is the best approach. It even makes a perverse kind of
sense to use the identity function for both parameters - grouping equal objects
in groups represented by their first occurrence.

[Lasse Reichstein Nielsen, 2016/02/25 16:18:10]

Done.

+      : _key = key, _value = value ?? _identity;

[nweiz, 2016/01/29 22:19:45]

Nit: put each of these on its own line
(https://www.dartlang.org/effective-dart/style/#do-format-constructor-initiali...).

[Lasse Reichstein Nielsen, 2016/02/25 16:18:10]

Done. Grudgingly.

+
+  // Helper function.
+  static _identity(x) => x;
+
+  Stream<Group<K, V>> bind(Stream<S> stream) async* {

[nweiz, 2016/01/29 22:19:44]

Last I heard, "async*" was really slow on the VM. Normally I'd be fine with it
because it's so much cleaner, but it worries me a little in a low-level utility
library.

[Lasse Reichstein Nielsen, 2016/02/01 12:43:20]

They should fix that then.
We have this syntax because it makes it *much* easier to write correct stream
code that handles pause and cancel properly.
We should ensure it's also fast. If it isn't, the bug *is* in the compiler.

[nweiz, 2016/02/02 02:29:13]

I agree for the most part (although in some situations they make it harder to
work with errors). But as far as I know, there's currently no momentum towards
fixing it, and it's not a major request from external users so it's not likely
to get on the VM team's radar in the near future. To some extent we have to make
do with the implementation we have, not the implementation we should have.

[Lasse Reichstein Nielsen, 2016/02/25 16:18:10]

I'd prefer to land this as written, then do a comparison and see if it's much
slower than the old-style version. If it is, we have a good case for
bug-reporting the VM performance. Changing it proactively means we won't have
that.

+    var controllers = new HashMap();

[nweiz, 2016/01/29 22:19:45]

"{}" would be a lot more idiomatic, and would make the dispatching of onDone
events have a deterministic order.

[Lasse Reichstein Nielsen, 2016/02/01 12:43:20]

But I don't want to have a deterministic order (or the overhead of a linked hash
map). The HashMap is the exact implementation that does what I need and want,
without adding any unused features.

A deterministic order just makes people depend on it. I could use a linked hash
map, but then I'd shuffle the values before using them :)

[nweiz, 2016/02/02 02:29:13]

I wrote a lot of Ruby code in the days when hashes were unordered by default,
and in my experience it was more common that people (including me!) didn't
realize the order wasn't deterministic and wrote tests towards whatever order
was coincidentally consistent in that version of the code. Then when some
unrelated piece of code changed later, it caused mysterious test failures
elsewhere.

For async stuff in particular, there's already so much nondeterminism that gives
rise to so many difficult-to-track-down, difficult-to-reproduce bugs, it seems
unkind to intentionally inject more.

[Lasse Reichstein Nielsen, 2016/02/25 16:18:10]

In this case, we do leak the order by closing the constructors in iteration
order, so that argument has merit.

I still think relying on ordering of independent async events is going to bite
you either way, so biting you early is arguably better.

This is exactly a case that HashMap exists for - a mapping where ordering isn't
important. I think I'll retain it.

+    try {
+      await for (S source in stream) {

[nweiz, 2016/01/29 22:19:45]

I'd use "var source" here and "var value" below.

[Lasse Reichstein Nielsen, 2016/02/25 16:18:10]

For value it's actually deliberate (but for source it's not necessary). We don't
enforce that _key or _value return the correct type (they could be returning
dynamic, and will be if the parameter was omitted), so this is a fail-early way
to catch incorrectly typed results in checked mode.
That's also why I introduce a variable for `value` - a type error on that line
very clearly describes the problem.

Now that `key` can be omitted too, I'll add `K` to the `key` variable too.

+        var key = _key(source);
+        var controller = controllers[key];
+        if (controller == null) {
+          controller = new StreamController<V>(sync: true);
+          controllers[key] = controller;
+          yield new Group<K, V>(key, controller.stream);
+        }
+        V value = _value(source);
+        controller.add(value);
+      }
+    } finally {
+      for (var controller in controllers.values) {
+        controller.close();
+      }
+    }
+  }
+}
+
+/// Scans a stream's events and combine a result from the available events.

[floitsch, 2016/01/29 19:05:03]

combines.

But the sentence could be better. I'm not sure what exactly it does (without
having looked at the code).

Seems to be: "Chains events of a stream by combining each element with the
result of the previous one".

In any case you really want to have examples (with or without code) in the doc.

[Lasse Reichstein Nielsen, 2016/02/25 16:18:10]

Reworded and example added.

+///
+/// Combines each event with the previous accumulator value
+/// and emits the result.
+///
+/// Errors in the source stream or when calling the combine function will
+/// be reported on the result stream, and will stop the process.
+class Scan<S, A> implements StreamTransformer<S, A> {

[floitsch, 2016/01/29 19:05:02]

"Scan looks like a bad name to me".

[nweiz, 2016/01/29 22:19:45]

How is this different than Stream.fold()?

[Lasse Reichstein Nielsen, 2016/02/25 16:18:10]

Stream.fold returns a single Future, this returns a Stream of the accumulator
values computed along the way. 
The relation is that Stream.fold(v, combine) is the same as Stream.transform(new
Scan(v, combine)).last

[Lasse Reichstein Nielsen, 2016/02/25 16:18:10]

Name was taking from the reactive framework.
Alternatives appreciated.

+  final A _initial;
+  final Function _combine;
+
+  /// Accumulates stream events using [combine] starting with [initial].
+  Scan(A initial, A combine(A accumulator, S source))
+      : _initial = initial, _combine = combine;
+
+  Stream<A> bind(Stream<S> stream) async* {
+    A accumulator = _initial;
+    await for (S source in stream) {
+      accumulator = _combine(accumulator, source);
+      yield accumulator;
+    }
+  }
+}
+
+/// Drops any event that happens shortly after another event.
+///
+/// If an event happens within a certain [Duration] of the previous event,
+/// the new event is dropped, but the dropped event will still cause
+/// following events to be dropped until the entire duration has passed without
+/// any events.
+///
+/// This differs from [Throttle] which also drops events, but which only
+/// restarts its timer at events that aren't dropped.
+class Debounce<S> implements StreamTransformer<S, S> {
+  final Duration _interval;
+
+  /// Drops events sooner than [interval] after any other event.
+  Debounce(Duration interval) : _interval = interval;

[nweiz, 2016/01/29 22:19:45]

I think this would be cleaner as "this._interval".

[Lasse Reichstein Nielsen, 2016/02/01 12:43:20]

I will not make a public parameter of a public method (or constructor) have a
private name.

+
+  Stream<S> bind(Stream<S> stream) async* {
+    var interval = Duration.ZERO;  // Avoid dropping the first event.

[nweiz, 2016/01/29 22:19:45]

This seems a little too clever to me. What about:

    // Don't start yet to avoid dropping the first event.
    var stopwatch = new Stopwatch();
    await for (var source in stream) {
      if (stopwatch.elapsed >= _interval) yield source;
      stopwatch.reset();
      stopwatch.start();
    }
    stopwatch.stop();

Same thing below.

[Lasse Reichstein Nielsen, 2016/02/01 12:43:20]

Wouldn't work - the comparison is in the wrong direction.
This will always discard the first event.

I prefer the current approach to a "bool isFirst = true;" flag.
(I really, really hate those - it's a sign that your programming language isn't
expressive enough to handle a fencepost problem).

[nweiz, 2016/02/02 02:29:13]

Hmm, true.
I'm not sure I've ever seen a language that was expressive enough to handle it
in full generality with anything I'd call grace.

+    var stopwatch = new Stopwatch()..start();
+    await for (var source in stream) {
+      if (stopwatch.elapsed >= interval) {
+        yield source;
+        interval = _interval;
+      }
+      stopwatch.reset();
+    }
+    stopwatch.stop();
+  }
+}
+
+/// Drops any event that happens shortly after another undropped event.
+///
+/// If an event happens within a certain [Duration] of the most recently
+/// output event, the new event is dropped.
+/// When the duration has passed after an event that wasn't dropped,
+/// the next event will be accepted.
+///
+/// This differs from [Debounce] which also drops events,

[floitsch, 2016/01/29 19:05:03]

from [Debounce] which restarts the timer after dropped events too.

[Lasse Reichstein Nielsen, 2016/02/25 16:18:10]

Done.

+/// but which restarts the timer after dropped events too.
+class Throttle<S> implements StreamTransformer<S, S> {
+  final Duration _interval;
+
+  /// Drops events sooner than [interval] after emitted events.
+  Throttle(Duration interval) : _interval = interval;
+
+  Stream<S> bind(Stream<S> stream) async* {
+    var interval = Duration.ZERO;  // Avoid dropping the first event.
+    var stopwatch = new Stopwatch()..start();
+    await for (var source in stream) {
+      if (stopwatch.elapsed >= interval) {
+        yield source;
+        stopwatch.reset();
+        interval = _interval;
+      }
+    }
+    stopwatch.stop();
+  }
+}
+
+/// Drops the last [count] events of a stream.
+///
+/// The resulting stream is [count] events behind the source stream.
+/// The first [count] events are buffered, and when further events arrive,
+/// they are buffered too, and the earlies events in the buffer are emitted.
+///
+/// The resulting stream ends [count] events before the source stream,
+/// effecively skipping the last [count] events.
+class SkipLast<S> implements StreamTransformer<S, S> {

[nweiz, 2016/01/29 22:19:45]

What's the use case for this?

[Lasse Reichstein Nielsen, 2016/02/01 12:43:20]

Honestly don't know - found it in the reactive framework and thought it sounded
simple to do, so let's do it.

[Lasse Reichstein Nielsen, 2016/02/25 16:18:10]

But since that's my best argument, I'll drop it for now and see if anybody
actually requests it.

+  /// Number of events dropped.
+  final int count;

[nweiz, 2016/01/29 22:19:45]

Why is this public? It doesn't match the parameters passed to other
transformers.

[Lasse Reichstein Nielsen, 2016/02/01 12:43:20]

Good point.

I think I decided to make it public because it means you can inspect the
transformer and see if it can be reused. That's probably not going to happen,
and if it does, it makes sense for the durations above too.

So, let's make it private.

+
+  /// Skip the [count] last events.
+  SkipLast(this.count);
+
+  Stream<S> bind(Stream<S> stream) async* {
+    Queue queue = new ListQueue(count + 1);

[nweiz, 2016/01/29 22:19:45]

Calling "new ListQueue()" explicitly reads very strange to me. I wouldn't
construct a specific implementation unless it would substantially improve
performance.

[Lasse Reichstein Nielsen, 2016/02/01 12:43:20]

The Queue constructor doesn't have an "initial capacity" parameter, so I need to
use the ListQueue constructor directly.
(Or not bother, which is obviously an option).

+    await for (S source in stream) {
+      queue.add(source);
+      if (queue.length > count) {
+        yield queue.removeFirst();
+      }
+    }
+  }
+}
+
+
+/**

[nweiz, 2016/01/29 22:19:45]

///

[Lasse Reichstein Nielsen, 2016/02/25 16:18:10]

Done.

+ * Groups events by a computed key.
+ *
+ * A key is extracted from incoming events, and a stream is created for each
+ * unique key. A value is computed from the event as well, and emitted on the
+ * corresponding stream.
+ *
+ * The returned stream emits a [Group] object for each distinct key seen
+ * by the transformation, and the values associated with the key are output
+ * on the [Group.values] stream.
+ */
+StreamTransformer<dynamic, Group> streamGroupBy(key(source), value(source)) =>

[nweiz, 2016/01/29 22:19:45]

It seems weird that both these functions *and* the StreamTransformer classes are
public. I'd just choose one (probably the functions).

Also, I'd call this "groupStreamBy" so that it reads a little more like prose.

[Lasse Reichstein Nielsen, 2016/02/01 12:43:20]

Until generic functions, that would prevent access to the the type parameters -
but I guess that's worth waiting for.
Good point.

+    new GroupBy(key, value);

[floitsch, 2016/01/29 19:05:03]

the resulting stream should have a "Future<Map> toMap()".

[Lasse Reichstein Nielsen, 2016/02/01 12:43:20]

Streams don't have that, and I *really* prefer not extending the stream
interface. I'd rather create a "collector" transformer to be the dual of the
groupBy.

+
+/// Scans a stream's events and combine a result from the available events.
+///
+/// Combines each event with the previous accumulator value
+/// and emits the result.
+///
+/// Errors in the source stream or when calling the combine function will
+/// be reported on the result stream, and will stop the process.
+StreamTransformer scanStream(initial, combine(value, source)) =>
+    new Scan(initial, combine);
+
+/// Drops any event that happens shortly after another undropped event.
+///
+/// If an event happens within a certain [Duration] of the most recently
+/// output event, the new event is dropped.
+/// When the duration has passed after an event that wasn't dropped,
+/// the next event will be accepted.
+///
+/// This differs from [Debounce] which also drops events,
+/// but which restarts the timer after dropped events too.
+StreamTransformer throttleStream(Duration duration) => new Throttle(duration);
+
+/// Drops any event that happens shortly after another event.
+///
+/// If an event happens within a certain [Duration] of the previous event,
+/// the new event is dropped, but the dropped event will still cause
+/// following events to be dropped until the entire duration has passed without
+/// any events.
+///
+/// This differs from [Throttle] which also drops events, but which only
+/// restarts its timer at events that aren't dropped.
+StreamTransformer debounceStream(Duration duration) => new Debounce(duration);
+
+/// Drops the last [count] events of a stream.
+StreamTransformer streamSkipLast(int count) => new SkipLast(count);
+
+/// Concatenates streams.
+///
+/// The events of each stream in [streams] are emitted, one stream at a time
+/// in iteration order. When the final stream in [streams] has been processed,
+/// the returned stream closes.
+Stream concatenateStreams(Iterable<Stream> streams) async* {

[nweiz, 2016/01/29 22:19:44]

This and mergeStreams aren't actually transformers, so they should probably go
in a different file. I think it's appropriate to put them in the top-level
lib/async.dart, but lib/src/stream_functions.dart would work as well.

[Lasse Reichstein Nielsen, 2016/02/25 16:18:10]

I've remove the rest, so it's just concatenateStreams now. I'll just make a
single file for that.

+  for (var stream in streams) {
+    yield* stream;
+  }
+}
+
+/// Returns a stream that emits the events of all the [streams].
+///
+/// The events from all the source streams are merged into one stream, with
+/// events from the same source stream occuring in their original order.
+Stream mergeStreams(Iterable<Stream> streams) {

[nweiz, 2016/01/29 22:19:45]

Isn't this the same as StreamGroup.merge?

[Lasse Reichstein Nielsen, 2016/02/01 12:43:20]

Most likely.

So, even if it's worth having as a top-level function (if you don't have a
StreamGroup, but just a bunch of streams, why would you look at StreamGroup for
the functionality), then it should at least reuse the implementation. 

[nweiz, 2016/02/02 02:29:13]

I'd prefer not to add a top-level forwarder; I'm not a fan of providing multiple
ways to do the same thing that users or style guides have to choose between. If
you do, though, we should find a way to do the same thing for other static
functions like StreamSplitter.split and StreamCompleter.fromFuture, for
completeness.

[Lasse Reichstein Nielsen, 2016/02/25 16:18:10]

Ok, I'll remove this one.

I'll remove the functions above too, except for concatenateStreams. Let's keep
just the classes.

(To be honest, I think the StreamTransformer interface was a mistake - it only
has a single function (there's a style guide rule against that), and
transformers should be reusable so they won't have any dynamic state anyway -
but we have it now, so let's embrace it and work towards making "new" optional
to reduce the clutter that way :)

+  var controller;
+  int activeSubscriptions = 0;
+  var subscriptions = [];
+  controller = new StreamController(sync: true,
+    onListen: () {
+      // Cache extracted functions instead of extacting for each stream.
+      var add = controller.add;
+      var addError = controller.addError;
+      for (var stream in streams) {
+        subscriptions.add(stream.listen(add,
+                                        onError: addError,
+                                        onDone: () {
+                                          activeSubscriptions--;
+                                          if (activeSubscriptions == 0) {
+                                            controller.close();
+                                          }
+                                        }));
+      }
+      activeSubscriptions = subscriptions.length;
+      if (activeSubscriptions == 0) {
+        controller.close();
+      }
+    },
+    onPause: () {
+      for (var subscription in subscriptions) subscription.pause();
+    },
+    onResume: () {
+      for (var subscription in subscriptions) subscription.resume();
+    },
+    onCancel: () {
+      return Future.wait(subscriptions.map((subscription) {
+        var result = subscription.cancel();
+        if (result is Future) return result;
+        return new Future.value();
+      }));
+  });
+  return controller.stream;
+}