Add groupBy to Stream.

tests/lib/async/stream_group_by_test.dart

+// Copyright (c) 2017, 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.
+
+library stream_group_by_test;
+
+import "dart:async";
+
+import "package:expect/expect.dart";
+import "package:async_helper/async_helper.dart";
+
+int len(x) => x.length;
+String wrap(x) => "[$x]";
+
+void main() {
+  asyncStart();
+  // groupBy.
+  test("splits", () async {
+    var grouped = stringStream.groupBy<int>(len);
+    var byLength = {};
+    await for (StreamGroup<int, String> group in grouped) {
+      byLength[group.key] = group.values.toList();
+    }
+    Expect.listEquals([1, 2, 4, 3], byLength.keys.toList());
+    expectCompletes(byLength[1], ["a", "b"]);
+    expectCompletes(byLength[2], ["ab"]);
+    expectCompletes(byLength[3], ["abe", "lea"]);
+    expectCompletes(byLength[4], ["abel", "bell", "able", "abba"]);
+  });
+
+  test("empty", () async {
+    var grouped = emptyStream.groupBy<int>(len);
+    var byLength = {};
+    await for (StreamGroup<int, String> group in grouped) {
+      byLength[group.key] = group.values.toList();
+    }
+    Expect.isTrue(byLength.isEmpty);
+  });
+
+  test("single group", () async {
+    var grouped = repeatStream(5, "x").groupBy<int>(len);
+    var byLength = {};
+    await for (StreamGroup<int, String> group in grouped) {
+      byLength[group.key] = group.values.toList();
+    }
+    Expect.listEquals([1], byLength.keys.toList());
+    expectCompletes(byLength[1], ["x", "x", "x", "x", "x"]);
+  });
+
+  test("with error", () async {
+    var grouped = stringErrorStream(3).groupBy<int>(len);
+    var byLength = {};
+    bool caught = false;
+    try {
+      await for (StreamGroup<int, String> group in grouped) {
+        byLength[group.key] = group.values.toList();
+      }
+    } catch (e) {
+      Expect.equals("BAD", e);
+      caught = true;
+    }
+    Expect.isTrue(caught);
+    Expect.listEquals([1, 2, 4], byLength.keys.toList());
+    expectCompletes(byLength[1], ["a", "b"]);
+    expectCompletes(byLength[2], ["ab"]);
+    expectCompletes(byLength[4], ["abel"]);
+  });
+
+  // groupValuesBy

[floitsch, 2017/05/02 12:23:53]

Nit. finish with ".".

[Lasse Reichstein Nielsen, 2017/05/04 11:05:20]

Every test below here removed. Problem solved!

+  test("splits", () async {

[floitsch, 2017/05/02 12:23:53]

There is already a test called "splits".

[Lasse Reichstein Nielsen, 2017/05/04 11:05:20]

Acknowledged.

+    var grouped = stringStream.groupValuesBy<int, String>(len, wrap);
+    var byLength = {};
+    await for (StreamGroup<int, String> group in grouped) {
+      byLength[group.key] = group.values.toList();
+    }
+    Expect.listEquals([1, 2, 4, 3], byLength.keys.toList());
+    expectCompletes(byLength[1], ["[a]", "[b]"]);
+    expectCompletes(byLength[2], ["[ab]"]);
+    expectCompletes(byLength[3], ["[abe]", "[lea]"]);
+    expectCompletes(byLength[4], ["[abel]", "[bell]", "[able]", "[abba]"]);
+  });
+
+  test("empty", () async {
+    var grouped = emptyStream.groupValuesBy<int>(len, wrap);
+    var byLength = {};
+    await for (StreamGroup<int, String> group in grouped) {
+      byLength[group.key] = group.values.toList();
+    }
+    Expect.isTrue(byLength.isEmpty);
+  });
+
+  test("single group", () async {
+    var grouped = repeatStream(5, "x").groupValuesBy<int>(len, wrap);
+    var byLength = {};
+    await for (StreamGroup<int, String> group in grouped) {
+      byLength[group.key] = group.values.toList();
+    }
+    Expect.listEquals([1], byLength.keys.toList());
+    expectCompletes(byLength[1], ["[x]", "[x]", "[x]", "[x]", "[x]"]);
+  });
+
+  test("with error", () async {
+    var grouped = stringErrorStream(3).groupValuesBy<int>(len, wrap);
+    var byLength = {};
+    bool caught = false;
+    try {
+      await for (StreamGroup<int, String> group in grouped) {
+        byLength[group.key] = group.values.toList();
+      }
+    } catch (e) {
+      Expect.equals("BAD", e);
+      caught = true;
+    }
+    Expect.isTrue(caught);
+    Expect.listEquals([1, 2, 4], byLength.keys.toList());
+    expectCompletes(byLength[1], ["[a]", "[b]"]);
+    expectCompletes(byLength[2], ["[ab]"]);
+    expectCompletes(byLength[4], ["[abel]"]);
+  });
+
+  asyncEnd();
+}
+
+expectCompletes(future, result) {
+  asyncStart();
+  future.then((v) {
+    if (result is List) {
+      Expect.listEquals(result, v);
+    } else {
+      Expect.equals(v, result);
+    }
+    asyncEnd();
+  }, onError: (e, s) {
+    Expect.fail("$e\n$s");
+  });
+}
+
+void test(name, func) {
+  asyncStart();
+  func().then((_) {
+    asyncEnd();
+  }, onError: (e, s) {
+    Expect.fail("$name: $e\n$s");
+  });
+}
+
+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;
+  }
+}