Add reactive-inspired stream transformers:

test/stream_transformers_test.dart

Index: test/stream_transformers_test.dart
diff --git a/test/stream_transformers_test.dart b/test/stream_transformers_test.dart
new file mode 100644
index 0000000000000000000000000000000000000000..97698d5694e9e0999eec156aeb17950405ec3682
--- /dev/null
+++ b/test/stream_transformers_test.dart
@@ -0,0 +1,220 @@
+// Copyright (c) 2016, the Dart project authors.  Please see the AUTHORS file

[nweiz, 2016/03/01 02:10:04]

Shouldn't these tests be deleted, since the individual classes each have their
own file now?

[Lasse Reichstein Nielsen, 2016/03/01 16:51:17]

Absolutely 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.
+
+// Test stream transformers.
+import "dart:async";
+
+import "package:async/async.dart";
+import "package:test/test.dart";
+
+var strings = const [
+  "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;
+
+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", () {
+    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 {
+      var grouped =
+         emptyStream.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, isEmpty);
+    });
+
+    test("single group", () async {
+      var grouped =
+         repeatStream(5, "x").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, [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;
+      }
+      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));
+    });
+  });
+
+
+  const ms = const Duration(milliseconds: 1);
+
+  group("unbounce", () {
+    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"),
+             ]);
+      });
+  });
+}