Add reactive-inspired stream transformers:

test/group_by_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";
+
+void main() {
+  test("splits", () async {
+    var stream = new Stream.fromIterable(
+        ["a", "ab", "b", "abel", "abe", "bell", "able", "abba", "lea"]);
+    var grouped = stream
+        .transform(new GroupBy<String, int, String>(key: (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("value transforms", () async {
+    var stream = new Stream.fromIterable(
+        ["a", "ab", "b", "abel", "abe", "bell", "able", "abba", "lea"]);
+    var grouped = stream.transform(new GroupBy<String, int, String>(
+        key: (x) => x.length, value: (s) => "<$s>"));
+    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 source", () async {
+    var emptyStream = new Stream<int>.empty();
+    var grouped = emptyStream.transform(new GroupBy<int, int, String>(
+        key: (_) => throw "key unreachable",
+        value: (_) => throw "value unreachable"));
+    expect(grouped.toList(), completion(isEmpty));
+  });
+
+  test("single group", () async {
+    var stream = new Stream<String>.fromIterable(["a", "b", "c", "d", "e"]);
+    var grouped = stream
+        .transform(new GroupBy<String, int, String>(key: (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(["a", "b", "c", "d", "e"]));
+  });
+
+  test("source error", () async {
+    var stream = new Stream<String>.fromIterable(
+        ["a", "ab", "b", "abel", "X", "abe", "bell", "able", "abba", "lea"]);
+    var errorStream = stream.map((v) => v == "X" ? throw "BAD" : v);
+    var grouped = errorStream
+        .transform(new GroupBy<String, int, String>(key: (x) => x.length));
+    var byLength = {};
+    int errorCount = 0;
+    await for (Result<Group<int, String>> result in
+              Result.captureStream(grouped)) {
+      if (result.isValue) {
+        var group = result.asValue.value;
+        byLength[group.key] = group.values.toList();
+      } else {
+        errorCount++;
+        expect(result.asError.error, "BAD");
+      }
+    }
+    expect(errorCount, 1);
+    // Doesn't stop at the error event.
+    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("key error", () async {
+    var stream = new Stream<String>.fromIterable(
+        ["a", "ab", "b", "abel", "X", "abe", "bell", "able", "abba", "lea"]);
+    badKey(x) => x == "X" ? throw "BAD" : x.length;
+    var grouped =
+        stream.transform(new GroupBy<String, int, String>(key: badKey));
+    var byLength = {};
+    int errorCount = 0;
+    await for (Result<Group<int, String>> result in
+              Result.captureStream(grouped)) {
+      if (result.isValue) {
+        var group = result.asValue.value;
+        byLength[group.key] = group.values.toList();
+      } else {
+        errorCount++;
+        expect(result.asError.error, "BAD");
+      }
+    }
+    expect(errorCount, 1);
+    // Doesn't stop at the error event.
+    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("value error", () async {
+    var stream = new Stream<String>.fromIterable(
+        ["a", "ab", "b", "abel", "XXX", "abe", "bell", "able", "abba", "lea"]);
+    badValue(x) => x == "XXX" ? throw "BAD" : x;
+    var grouped = stream.transform(new GroupBy<String, int, String>(
+        key: (x) => x.length, value: badValue));
+    var byLength = {};
+    await for (Group<int, String> group in grouped) {
+      byLength[group.key] = Result.captureStream(group.values).toList();
+    }
+    // Doesn't stop at the error event.
+    expect(byLength.keys.toList(), [1, 2, 4, 3]);
+    Result r(v) => new Result<String>.value(v);
+    expect(byLength[1], completion([r("a"), r("b")]));
+    expect(byLength[2], completion([r("ab")]));
+    /// The length-3 stream starts with an error "BAD" and then two values.
+    expect(byLength[3].then((v) => v.first.isError),
+           completion(true));
+    expect(byLength[3].then((v) => v.first.asError.error),
+           completion("BAD"));
+    expect(byLength[3].then((v) => v.sublist(1)),
+           completion([r("abe"), r("lea")]));
+    expect(byLength[4],
+           completion([r("abel"), r("bell"), r("able"), r("abba")]));
+  });
+}

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

Test pausing and canceling behavior as well.

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

Acknowledged.