Add reactive-inspired stream transformers:

test/stream_transformers_test.dart

+// Copyright (c) 2016, 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.
+
+// Test stream transformers.
+import "dart:async";
+
+import "package:async/async.dart";
+import "package:test/test.dart";
+
+var strings = const [

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

I find it a lot easier to read tests when they include all their data inline or
in setUp/tearDown, rather than using externally-defined data that I need to look
up. It does involve more duplication, but it makes it easier to look at the test
and figure out what's going on, or to change its data later without affecting
other tests.

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

Done.

+  "a",
+  "ab",
+  "b",
+  "abel",
+  "abe",
+  "bell",
+  "able",
+  "abba",
+  "lea"
+];
+
+Stream<String> get stringStream async* {
+  for (var string in strings) {
+    yield string;
+  }
+}
+
+Stream get emptyStream async* {}
+
+Stream repeatStream(int count, value) async* {
+  for (var i = 0; i < count; i++) {
+    yield value;
+  }
+}
+
+// Just some valid stack trace.
+var stack = StackTrace.current;

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

It would be nice to avoid the tests depending on 1.14, at least until it gets a
stable release.

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

I believe 1.14 has a stable release now.

+
+Stream<String> stringErrorStream(int errorAfter) async* {
+  for (int i = 0; i < strings.length; i++) {
+    yield strings[i];
+    if (i == errorAfter) {
+      // Emit error, but continue afterwards.
+      yield* new Future.error("BAD", stack).asStream();
+    }
+  }
+}
+
+Stream intStream(int count, [int start = 0]) async* {
+  for (int i = 0; i < count; i++) {
+    yield start++;
+  }
+}
+
+Stream timerStream(int count, Duration interval) async* {
+  for (int i = 0; i < count; i++) {
+    await new Future.delayed(interval);
+    yield i;
+  }
+}
+
+void main() {
+  group("groupBy", () {

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

Test the value() function as well. Test what happens when key() or value()
throws an error.

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

Done.

+    test("splits", () async {
+      var grouped =
+         stringStream.transform(new GroupBy<int, String>((x) => x.length));
+      var byLength = {};
+      await for (Group<int,String> group in grouped) {
+        byLength[group.key] = group.values.toList();
+      }
+      expect(byLength.keys.toList(), [1, 2, 4, 3]);
+      expect(byLength[1], completion(["a", "b"]));
+      expect(byLength[2], completion(["ab"]));
+      expect(byLength[3], completion(["abe", "lea"]));
+      expect(byLength[4], completion(["abel", "bell", "able", "abba"]));
+    });
+
+    test("empty", () async {

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

group + test should give a sentence.

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

No groups any longer.

+      var grouped =
+         emptyStream.transform(new GroupBy<int, String>((x) => x.length));
+      var byLength = {};
+      await for (Group<int,String> group in grouped) {

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

It would be cleaner to just write expect(grouped.toList(), completion(isEmpty)).

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

Done.

+        byLength[group.key] = group.values.toList();
+      }
+      expect(byLength, isEmpty);
+    });
+
+    test("single group", () async {
+      var grouped =
+         repeatStream(5, "x").transform(new GroupBy<int, String>((x) => x.length));

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

long line.

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

Done.

+      var byLength = {};
+      await for (Group<int,String> group in grouped) {
+        byLength[group.key] = group.values.toList();
+      }
+      expect(byLength.keys, [1]);
+      expect(byLength[1], completion(["x", "x", "x", "x", "x"]));
+    });
+
+    test("with error", () async {
+      var grouped = stringErrorStream(3).
+          transform(new GroupBy<int, String>((x) => x.length));
+      var byLength = {};
+      bool caught = false;
+      try {
+        await for (Group<int,String> group in grouped) {
+          byLength[group.key] = group.values.toList();
+        }
+      } catch (e) {
+        expect(e, "BAD");
+        caught = true;
+      }

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

It seems strange, for this and other transformers, that any error stops the
stream dead. That's not consistent with how the dart:async transformers work,
and it's less flexible than the alternative—you could write a "close after
error" transformer, but not a "continue after error" one.

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

It is a problem that our async*/await for primitives can't handle errors in the
source. However, that does mean that "multiple errors on the same stream" is
really not a thing any more - new code (using await for) won't be able to handle
it anyway.

So, when writing new stream functionality, I think it's acceptable to die on the
first error (as long as you still do that properly).

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

I'm pretty skeptical of this. There are still plenty of streams that
meaningfully emit multiple errors. Often those exposed by classes representing
long-running processes do so, for example. Making a bunch of transformers with
inconsistent behavior around those errors will make it much harder to work with
those streams, because some (potentially unknown) subset of transformers will be
unusable, but won't obviously fail to work at first glance.

If we were comfortable altering the core library to forbid any stream to emit
multiple events, I'd be okay with this. But given that that's off the table, I
think we have to make new transformers—at least ones as general-purpose as
these—support the existing model.

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

Reasonable. I'll try rewriting it to accept errors and forward all source and
key errors to the main stream, and errors from calling the value function to the
key-group stream. That also means not using async*.
(And actually having to handle, and think about, pause and cancel explicitly).

+      expect(caught, isTrue);
+      expect(byLength.keys.toList(), [1, 2, 4]);
+      expect(byLength[1], completion(["a", "b"]));
+      expect(byLength[2], completion(["ab"]));
+      expect(byLength[4], completion(["abel"]));
+    });
+  });
+
+  group("Scan", () {
+    test("adds", () {
+      var scanned = intStream(5, 2)
+          .transform(new Scan<int, int>(0, (a, v) => a + v));
+      expect(scanned.toList(), completion([2, 5, 9, 14, 20]));
+    });
+    test("subtracts", () {
+      // This will not work if parameters are swapped.
+      var scanned = intStream(5, 2)
+          .transform(new Scan<int, int>(0, (a, v) => a - v));
+      expect(scanned.toList(), completion([-2, -5, -9, -14, -20]));
+    });
+    test("types", () {
+      var scanned = intStream(5, 2)
+          .transform(new Scan<int, String>("", (a, v) => "$a$v"));
+      expect(scanned.toList(), completion(["2", "23", "234", "2345", "23456"]));
+    });
+    test("types 2", () {
+      var scan = new Scan<int, String>("", (a, v) => "$a$v");
+      var scanned = scan.bind(intStream(5, 2));
+      expect(scanned is Stream<String>, isTrue);
+      expect(scanned.toList(), completion(["2", "23", "234", "2345", "23456"]));
+    });
+    test("error", () async {
+      var calls = 0;
+      var scanned = stringErrorStream(3)
+          .transform(new Scan<String, int>(0, (a, b) => ++calls));
+      Future list = scanned.toList();
+      Future listDone = list.catchError((_) => calls);
+      expect(list, throwsA("BAD"));
+      expect(listDone, completion(4));
+    });
+  });
+

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

Extra newline.

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

Rewritten.

+
+  const ms = const Duration(milliseconds: 1);

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

Making this a method-level const declaration reads very strange to me.

Personally, I'd just manually write out "new Duration()" below, but if not I
think it would be clearer to move this to the top level.

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

Is it because it's const, because it could just as well be "var/new" instead?

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

Const makes it weirder, but it would still look weird as var/new too.

+
+  group("unbounce", () {

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

This and throttle need more tests. They should test error behavior, that the
right events got dropped, and that events are *not* dropped if there's enough
time between them.

You can use the clock and fake_async packages to mock out the Stopwatch.

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

The clock and fake_async packages require you to use other methods than "new
Timer"/"new Stopwatch", so that you can mock them for testing, right?

It does make sense, even if I am loathe to change code for testing - it seems
somehow dishonest to change the code that you are testing to match the test, it
should be the other way around.

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

I suppose ideally the Zone API would mock them out itself.

I'm generally not a big fan of going to great lengths to mock stuff out, but
timing stuff is such a source of flakiness when it's unmocked and is so easy
when it is that I make an exception for it.

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

Ideally you could replace the core library with something that changed some of
the classes. That's a better level of abstraction than making every static
function or class interceptable using the already heavily overloaded zones.
First class libraries are a hard sell, though.

+    test("keeps restarting timer", () async {
+      var debounced = timerStream(18, ms * 10)
+          .transform(new Debounce<int>(ms * 100));
+      expect(debounced.length, completion(1));
+    });
+  });
+
+  group("throttle", () {
+    // Very hard to test safely because events may be delayed.
+    test("keeps restarting timer", () async {
+      var throttled = timerStream(16, ms * 10)
+          .transform(new Throttle<int>(ms * 50));
+      var length = throttled.length;
+      expect(length, completion(lessThan(6)));
+      expect(length, completion(greaterThan(1)));
+    });
+  });
+
+  group("merge", () {
+    test("none", () async {
+      var stream = mergeStreams([]);
+      int count = 0;
+      await for (var event in stream) {
+        count++;
+      }
+      expect(count, 0);
+    });
+
+    test("single", () async {
+      var stream = mergeStreams([intStream(5)]);
+      expect(stream.toList(), completion([0, 1, 2, 3, 4]));
+    });
+
+    test("multiple", () async {
+      var stream = mergeStreams([intStream(5), stringStream]);
+      var events = await stream.toList();
+      expect(events.where((x) => x is int), [0, 1, 2, 3, 4]);
+      expect(events.where((x) => x is String), strings);
+      expect(events.where((x) => x is! int && x is! String), isEmpty);
+    });
+
+    test("error", () async {
+      var stream = mergeStreams([stringErrorStream(3), intStream(6)]);
+      var events = await Result.captureStream(stream).toList();
+      expect(events.where((Result x) => x.isValue && x.asValue.value is int)
+                   .map((ValueResult x) => x.value),
+             [0, 1, 2, 3, 4, 5]);
+      expect(events.where((x) => x.isError || x.asValue.value is! int),
+             [new Result.value("a"),
+              new Result.value("ab"),
+              new Result.value("b"),
+              new Result.value("abel"),
+              new Result.error("BAD", stack),
+              new Result.value("abe"),
+              new Result.value("bell"),
+              new Result.value("able"),
+              new Result.value("abba"),
+              new Result.value("lea"),
+             ]);
+      });
+  });
+}