Implement merge and mergeAll (dart-lang/stream_transform#7) Closes dart-lang/stream_transform#4 merge and mergeAll combine/interleave streams. Handles both broadcast and single-subscription streams. Add implementation of containsAllInOrder in test/util - once this is submitted in package:matcher it can be removed here. Submitting here since the other package is blocked. Remove -dev from pubspec for publishing.
diff --git a/pkgs/stream_transform/CHANGELOG.md b/pkgs/stream_transform/CHANGELOG.md index 1d323aa..c820833 100644 --- a/pkgs/stream_transform/CHANGELOG.md +++ b/pkgs/stream_transform/CHANGELOG.md
@@ -1,6 +1,7 @@ ## 0.0.2 - Add `concat`: Appends streams in series +- Add `merge` and `mergeAll`: Interleaves streams ## 0.0.1
diff --git a/pkgs/stream_transform/README.md b/pkgs/stream_transform/README.md index ac32362..b5ffaa1 100644 --- a/pkgs/stream_transform/README.md +++ b/pkgs/stream_transform/README.md
@@ -14,3 +14,7 @@ Prevents a source stream from emitting too frequently by dropping or collecting values that occur within a given duration. + +# merge, mergeAll + +Interleaves events from multiple streams into a single stream.
diff --git a/pkgs/stream_transform/lib/src/merge.dart b/pkgs/stream_transform/lib/src/merge.dart new file mode 100644 index 0000000..d2e53a6 --- /dev/null +++ b/pkgs/stream_transform/lib/src/merge.dart
@@ -0,0 +1,71 @@ +// 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. + +import 'dart:async'; + +/// Emits values from the source stream and [other] in any order as they arrive. +/// +/// If the source stream is a broadcast stream, the result stream will be as +/// well, regardless of the type of stream [other] is. +StreamTransformer<T, T> merge<T>(Stream<T> other) => new _Merge<T>([other]); + +/// Emits values from the source stream and all streams in [others] in any order +/// as they arrive. +/// +/// If the source stream is a broadcast stream, the result stream will be as +/// well, regardless of the types of streams in [others]. +StreamTransformer<T, T> mergeAll<T>(List<Stream<T>> others) => + new _Merge<T>(others); + +class _Merge<T> implements StreamTransformer<T, T> { + final List<Stream<T>> _others; + + _Merge(this._others); + + @override + Stream<T> bind(Stream<T> first) { + StreamController<T> controller; + if (first.isBroadcast) { + controller = new StreamController<T>.broadcast(); + } else { + controller = new StreamController<T>(); + } + List<StreamSubscription> subscriptions; + List<Stream<T>> allStreams = [first]..addAll(_others); + var activeStreamCount = 0; + + controller.onListen = () { + if (subscriptions != null) return; + subscriptions = allStreams.map((stream) { + activeStreamCount++; + return stream.listen(controller.add, onError: controller.addError, + onDone: () { + if (--activeStreamCount <= 0) controller.close(); + }); + }).toList(); + }; + + // Forward methods from listener + if (!first.isBroadcast) { + controller.onPause = () { + for (var subscription in subscriptions) { + subscription.pause(); + } + }; + controller.onResume = () { + for (var subscription in subscriptions) { + subscription.resume(); + } + }; + controller.onCancel = + () => Future.wait(subscriptions.map((s) => s.cancel())); + } else { + controller.onCancel = () { + if (controller?.hasListener ?? false) return new Future.value(null); + return Future.wait(subscriptions.map((s) => s.cancel())); + }; + } + return controller.stream; + } +}
diff --git a/pkgs/stream_transform/lib/stream_transform.dart b/pkgs/stream_transform/lib/stream_transform.dart index c128e4b..de66dbe 100644 --- a/pkgs/stream_transform/lib/stream_transform.dart +++ b/pkgs/stream_transform/lib/stream_transform.dart
@@ -3,5 +3,6 @@ // BSD-style license that can be found in the LICENSE file. export 'src/buffer.dart'; -export 'src/debounce.dart'; export 'src/concat.dart'; +export 'src/debounce.dart'; +export 'src/merge.dart';
diff --git a/pkgs/stream_transform/pubspec.yaml b/pkgs/stream_transform/pubspec.yaml index 7352049..3d096dc 100644 --- a/pkgs/stream_transform/pubspec.yaml +++ b/pkgs/stream_transform/pubspec.yaml
@@ -2,7 +2,7 @@ description: A collection of utilities to transform and manipulate streams. author: Dart Team <misc@dartlang.org> homepage: https://www.github.com/dart-lang/stream_transform -version: 0.0.2-dev +version: 0.0.2 environment: sdk: ">=1.22.0 <2.0.0"
diff --git a/pkgs/stream_transform/test/merge_test.dart b/pkgs/stream_transform/test/merge_test.dart new file mode 100644 index 0000000..2199ecb --- /dev/null +++ b/pkgs/stream_transform/test/merge_test.dart
@@ -0,0 +1,104 @@ +import 'dart:async'; + +import 'package:test/test.dart'; + +import 'package:stream_transform/stream_transform.dart'; + +import 'util/matchers.dart'; + +void main() { + group('merge', () { + test('includes all values', () async { + var first = new Stream.fromIterable([1, 2, 3]); + var second = new Stream.fromIterable([4, 5, 6]); + var allValues = await first.transform(merge(second)).toList(); + expect(allValues, containsAllInOrder([1, 2, 3])); + expect(allValues, containsAllInOrder([4, 5, 6])); + expect(allValues, hasLength(6)); + }); + + test('cancels both sources', () async { + var firstCanceled = false; + var first = new StreamController() + ..onCancel = () { + firstCanceled = true; + }; + var secondCanceled = false; + var second = new StreamController() + ..onCancel = () { + secondCanceled = true; + }; + var subscription = + first.stream.transform(merge(second.stream)).listen((_) {}); + await subscription.cancel(); + expect(firstCanceled, true); + expect(secondCanceled, true); + }); + }); + + group('mergeAll', () { + test('includes all values', () async { + var first = new Stream.fromIterable([1, 2, 3]); + var second = new Stream.fromIterable([4, 5, 6]); + var third = new Stream.fromIterable([7, 8, 9]); + var allValues = await first.transform(mergeAll([second, third])).toList(); + expect(allValues, containsAllInOrder([1, 2, 3])); + expect(allValues, containsAllInOrder([4, 5, 6])); + expect(allValues, containsAllInOrder([7, 8, 9])); + expect(allValues, hasLength(9)); + }); + + test('handles mix of broadcast and single-subscription', () async { + var firstCanceled = false; + var first = new StreamController.broadcast() + ..onCancel = () { + firstCanceled = true; + }; + var secondBroadcastCanceled = false; + var secondBroadcast = new StreamController.broadcast() + ..onCancel = () { + secondBroadcastCanceled = true; + }; + var secondSingleCanceled = false; + var secondSingle = new StreamController() + ..onCancel = () { + secondSingleCanceled = true; + }; + + var merged = first.stream + .transform(mergeAll([secondBroadcast.stream, secondSingle.stream])); + + var firstListenerValues = []; + var secondListenerValues = []; + + var firstSubscription = merged.listen(firstListenerValues.add); + var secondSubscription = merged.listen(secondListenerValues.add); + + first.add(1); + secondBroadcast.add(2); + secondSingle.add(3); + + await new Future(() {}); + await firstSubscription.cancel(); + + expect(firstCanceled, false); + expect(secondBroadcastCanceled, false); + expect(secondSingleCanceled, false); + + first.add(4); + secondBroadcast.add(5); + secondSingle.add(6); + + await new Future(() {}); + await secondSubscription.cancel(); + + await new Future(() {}); + expect(firstCanceled, true); + expect(secondBroadcastCanceled, true); + expect(secondSingleCanceled, true); + + expect(firstListenerValues, [1, 2, 3]); + expect(secondListenerValues, [1, 2, 3, 4, 5, 6]); + }); + }); +}
diff --git a/pkgs/stream_transform/test/util/matchers.dart b/pkgs/stream_transform/test/util/matchers.dart new file mode 100644 index 0000000..4028b73 --- /dev/null +++ b/pkgs/stream_transform/test/util/matchers.dart
@@ -0,0 +1,42 @@ +import 'package:test/test.dart'; + +/// Matches [Iterable]s which contain an element matching every value in +/// [expected] in the same order, but may contain additional values interleaved +/// throughout. +Matcher containsAllInOrder(Iterable expected) => + new _ContainsAllInOrder(expected); + +class _ContainsAllInOrder implements Matcher { + final Iterable _expected; + + _ContainsAllInOrder(this._expected); + + String _test(item, Map matchState) { + if (item is! Iterable) return 'not iterable'; + var matchers = _expected.map(wrapMatcher).toList(); + var matcherIndex = 0; + for (var value in item) { + if (matchers[matcherIndex].matches(value, matchState)) matcherIndex++; + if (matcherIndex == matchers.length) return null; + } + return new StringDescription() + .add('did not find a value matching ') + .addDescriptionOf(matchers[matcherIndex]) + .add(' following expected prior values') + .toString(); + } + + @override + bool matches(item, Map matchState) => _test(item, matchState) == null; + + @override + Description describe(Description description) => description + .add('contains in order(') + .addDescriptionOf(_expected) + .add(')'); + + @override + Description describeMismatch(item, Description mismatchDescription, + Map matchState, bool verbose) => + mismatchDescription.add(_test(item, matchState)); +}