Add package with iterable and stream "zipping" functionality.

pkg/sequence_zip/test/stream_test.dart

+// Copyright (c) 2013, 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.
+
+import "dart:async";
+import "package:sequence_zip/stream_zip.dart";
+import "package:unittest/unittest.dart";
+
+/// Create an error with the same values as [base], except that it throwsA
+/// when seeing the value [errorValue].
+Stream streamError(Stream base, int errorValue) {
+  return base.map((x) => (x == errorValue) ? throw "BAD" : x);
+}
+
+/// Make a [Stream] from an [Iterable] by adding events to a stream controller
+/// at periodic intervals.
+Stream mks(Iterable iterable) {
+  Iterator iterator = iterable.iterator;
+  StreamController controller = new StreamController();
+  // Some varying time between 3 and 10 ms.
+  int ms = ((++ctr) * 5) % 7 + 3;
+  new Timer.periodic(new Duration(milliseconds: ms), (Timer timer) {
+    if (iterator.moveNext()) {
+      controller.add(iterator.current);
+    } else {
+      controller.close();
+      timer.cancel();
+    }
+  });
+  return controller.stream;
+}
+
+/// Counter used to give varying delays for streams.
+int ctr = 0;
+
+main() {
+  // Test that zipping [streams] gives the results iterated by [expectedData].
+  testZip(Iterable streams, Iterable expectedData) {
+    List data = [];
+    Stream zip = new StreamZip(streams);
+    zip.listen(data.add, onDone: expectAsync0(() {
+      expect(data, equals(expectedData));
+    }));
+  }
+
+  test("Basic", () {
+    testZip([mks([1, 2, 3]), mks([4, 5, 6]), mks([7, 8, 9])],
+            [[1, 4, 7], [2, 5, 8], [3, 6, 9]]);
+  });
+
+  test("Uneven length 1", () {
+    testZip([mks([1, 2, 3, 99, 100]), mks([4, 5, 6]), mks([7, 8, 9])],
+            [[1, 4, 7], [2, 5, 8], [3, 6, 9]]);
+  });
+
+  test("Uneven length 2", () {
+    testZip([mks([1, 2, 3]), mks([4, 5, 6, 99, 100]), mks([7, 8, 9])],
+            [[1, 4, 7], [2, 5, 8], [3, 6, 9]]);
+  });
+
+  test("Uneven length 3", () {
+    testZip([mks([1, 2, 3]), mks([4, 5, 6]), mks([7, 8, 9, 99, 100])],
+            [[1, 4, 7], [2, 5, 8], [3, 6, 9]]);
+  });
+
+  test("Uneven length 4", () {
+    testZip([mks([1, 2, 3, 98]), mks([4, 5, 6]), mks([7, 8, 9, 99, 100])],
+            [[1, 4, 7], [2, 5, 8], [3, 6, 9]]);
+  });
+
+  test("Empty 1", () {
+    testZip([mks([]), mks([4, 5, 6]), mks([7, 8, 9])], []);
+  });
+
+  test("Empty 2", () {
+    testZip([mks([1, 2, 3]), mks([]), mks([7, 8, 9])], []);
+  });
+
+  test("Empty 3", () {
+    testZip([mks([1, 2, 3]), mks([4, 5, 6]), mks([])], []);
+  });
+
+  test("Empty source", () {
+    testZip([], []);
+  });
+
+  test("Single Source", () {
+    testZip([mks([1, 2, 3])], [[1], [2], [3]]);
+  });
+
+  test("Other-streams", () {
+    Stream st1 = mks([1, 2, 3, 4, 5, 6]).where((x) => x < 4);
+    Stream st2 = new Stream.periodic(const Duration(milliseconds: 5),

[floitsch, 2013/07/04 14:31:22]

This means that the test won't work in d8.

[Lasse Reichstein Nielsen, 2013/07/09 06:16:00]

I'll mark it as failing.

+                                     (x) => x + 4).take(3);
+    StreamController c = new StreamController.broadcast();
+    Stream st3 = c.stream;
+    testZip([st1, st2, st3],
+            [[1, 4, 7], [2, 5, 8], [3, 6, 9]]);
+    c..add(7)..add(8)..add(9)..close();
+  });
+
+  test("Error 1", () {
+    expect(new StreamZip([streamError(mks([1, 2, 3]), 2),
+                          mks([4, 5, 6]),
+                          mks([7, 8, 9])]).toList(),
+           throwsA(equals("BAD")));
+  });
+
+  test("Error 2", () {
+    expect(new StreamZip([mks([1, 2, 3]),
+                          streamError(mks([4, 5, 6]), 5),
+                          mks([7, 8, 9])]).toList(),
+           throwsA(equals("BAD")));
+  });
+
+  test("Error 3", () {
+    expect(new StreamZip([mks([1, 2, 3]),
+                          mks([4, 5, 6]),
+                          streamError(mks([7, 8, 9]), 8)]).toList(),
+           throwsA(equals("BAD")));
+  });
+
+  test("Error at end", () {
+    expect(new StreamZip([mks([1, 2, 3]),
+                          streamError(mks([4, 5, 6]), 6),
+                          mks([7, 8, 9])]).toList(),
+           throwsA(equals("BAD")));
+  });
+
+  test("Error before first end", () {
+    // StreamControllers' streams with no "close" called will never be done,
+    // so the fourth event of the first stream is guaranteed to come first.
+    expect(new StreamZip(
+                [streamError(mks([1, 2, 3, 4]), 4),
+                 (new StreamController()..add(4)..add(5)..add(6)).stream,
+                 (new StreamController()..add(7)..add(8)..add(9)).stream]
+               ).toList(),
+           throwsA(equals("BAD")));
+  });
+
+  test("Error after first end", () {
+    StreamController controller = new StreamController();
+    controller..add(7)..add(8)..add(9);
+    testZip([mks([1, 2, 3]),
+             mks([4, 5, 6]),
+             controller.stream],
+           [[1, 4, 7], [2, 5, 8], [3, 6, 9]]);
+    // This comes after the first three events in all cases, since they
+    // use durations no greater than 10 ms.
+    new Timer(const Duration(milliseconds: 100), () {
+      controller.addError("BAD");
+    });
+  });
+}