| // 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:async/async.dart"; |
| import "package:test/test.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, error) { |
| return base.map((x) => (x == errorValue) ? throw error : 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<Stream> 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), (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, "BAD-1"), |
| mks([4, 5, 6]), |
| mks([7, 8, 9]) |
| ]).toList(), |
| throwsA(equals("BAD-1"))); |
| }); |
| |
| test("Error 2", () { |
| expect( |
| new StreamZip([ |
| mks([1, 2, 3]), |
| streamError(mks([4, 5, 6]), 5, "BAD-2"), |
| mks([7, 8, 9]) |
| ]).toList(), |
| throwsA(equals("BAD-2"))); |
| }); |
| |
| test("Error 3", () { |
| expect( |
| new StreamZip([ |
| mks([1, 2, 3]), |
| mks([4, 5, 6]), |
| streamError(mks([7, 8, 9]), 8, "BAD-3") |
| ]).toList(), |
| throwsA(equals("BAD-3"))); |
| }); |
| |
| test("Error at end", () { |
| expect( |
| new StreamZip([ |
| mks([1, 2, 3]), |
| streamError(mks([4, 5, 6]), 6, "BAD-4"), |
| mks([7, 8, 9]) |
| ]).toList(), |
| throwsA(equals("BAD-4"))); |
| }); |
| |
| 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, "BAD-5"), |
| (new StreamController()..add(4)..add(5)..add(6)).stream, |
| (new StreamController()..add(7)..add(8)..add(9)).stream |
| ]).toList(), |
| throwsA(equals("BAD-5"))); |
| }); |
| |
| test("Error after first end", () { |
| StreamController controller = new StreamController(); |
| controller..add(7)..add(8)..add(9); |
| // Transformer that puts error into controller when one of the first two |
| // streams have sent a done event. |
| StreamTransformer trans = |
| new StreamTransformer.fromHandlers(handleDone: (EventSink s) { |
| Timer.run(() { |
| controller.addError("BAD-6"); |
| }); |
| s.close(); |
| }); |
| testZip([ |
| mks([1, 2, 3]).transform(trans), |
| mks([4, 5, 6]).transform(trans), |
| controller.stream |
| ], [ |
| [1, 4, 7], |
| [2, 5, 8], |
| [3, 6, 9] |
| ]); |
| }); |
| |
| test("Pause/Resume", () { |
| int sc1p = 0; |
| StreamController c1 = new StreamController(onPause: () { |
| sc1p++; |
| }, onResume: () { |
| sc1p--; |
| }); |
| |
| int sc2p = 0; |
| StreamController c2 = new StreamController(onPause: () { |
| sc2p++; |
| }, onResume: () { |
| sc2p--; |
| }); |
| |
| var done = expectAsync0(() { |
| expect(sc1p, equals(1)); |
| expect(sc2p, equals(0)); |
| }); // Call to complete test. |
| |
| Stream zip = new StreamZip([c1.stream, c2.stream]); |
| |
| const ms25 = const Duration(milliseconds: 25); |
| |
| // StreamIterator uses pause and resume to control flow. |
| StreamIterator it = new StreamIterator(zip); |
| |
| it.moveNext().then((hasMore) { |
| expect(hasMore, isTrue); |
| expect(it.current, equals([1, 2])); |
| return it.moveNext(); |
| }).then((hasMore) { |
| expect(hasMore, isTrue); |
| expect(it.current, equals([3, 4])); |
| c2.add(6); |
| return it.moveNext(); |
| }).then((hasMore) { |
| expect(hasMore, isTrue); |
| expect(it.current, equals([5, 6])); |
| new Future.delayed(ms25).then((_) { |
| c2.add(8); |
| }); |
| return it.moveNext(); |
| }).then((hasMore) { |
| expect(hasMore, isTrue); |
| expect(it.current, equals([7, 8])); |
| c2.add(9); |
| return it.moveNext(); |
| }).then((hasMore) { |
| expect(hasMore, isFalse); |
| done(); |
| }); |
| |
| c1 |
| ..add(1) |
| ..add(3) |
| ..add(5) |
| ..add(7) |
| ..close(); |
| c2..add(2)..add(4); |
| }); |
| |
| test("pause-resume2", () { |
| var s1 = new Stream.fromIterable([0, 2, 4, 6, 8]); |
| var s2 = new Stream.fromIterable([1, 3, 5, 7]); |
| var sz = new StreamZip([s1, s2]); |
| int ctr = 0; |
| StreamSubscription sub; |
| sub = sz.listen(expectAsync1((v) { |
| expect(v, equals([ctr * 2, ctr * 2 + 1])); |
| if (ctr == 1) { |
| sub.pause(new Future.delayed(const Duration(milliseconds: 25))); |
| } else if (ctr == 2) { |
| sub.pause(); |
| new Future.delayed(const Duration(milliseconds: 25)).then((_) { |
| sub.resume(); |
| }); |
| } |
| ctr++; |
| }, count: 4)); |
| }); |
| } |
