Add `combineLatest` (dart-lang/stream_transform#65) Closes dart-lang/stream_transform#63 Uses a manual `StreamTransformer` implementation based closely on the `merge` implementation. This can't be implemented on top of `fromHandlers` or normal stream methods because the events emitted aren't triggered by a single source.
diff --git a/pkgs/stream_transform/CHANGELOG.md b/pkgs/stream_transform/CHANGELOG.md index 371889a..1a2ce66 100644 --- a/pkgs/stream_transform/CHANGELOG.md +++ b/pkgs/stream_transform/CHANGELOG.md
@@ -4,6 +4,7 @@ behavior changes for synchronous callbacks. **Potential breaking change** In the unlikely situation where `scan` was used to produce a `Stream<Future>` inference may now fail and require explicit generic type arguments. +- Add `combineLatest`. ## 0.0.15
diff --git a/pkgs/stream_transform/README.md b/pkgs/stream_transform/README.md index 89ce15d..7e556cb 100644 --- a/pkgs/stream_transform/README.md +++ b/pkgs/stream_transform/README.md
@@ -19,6 +19,11 @@ Collects values from a source stream until a `trigger` stream fires and the collected values are emitted. +# combineLatest + +Combine the most recent event from two streams through a callback and emit the +result. + # debounce, debounceBuffer Prevents a source stream from emitting too frequently by dropping or collecting
diff --git a/pkgs/stream_transform/lib/src/combine_latest.dart b/pkgs/stream_transform/lib/src/combine_latest.dart new file mode 100644 index 0000000..1fe7d85 --- /dev/null +++ b/pkgs/stream_transform/lib/src/combine_latest.dart
@@ -0,0 +1,146 @@ +// Copyright (c) 2019, 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'; + +/// Combine the latest value from the source stream with the latest value from +/// [other] using [combine]. +/// +/// No event will be emitted from the result stream until both the source stream +/// and [other] have each emitted at least one event. Once both streams have +/// emitted at least one event, the result stream will emit any time either +/// input stream emits. +/// +/// For example: +/// source.transform(combineLatest(other, (a, b) => a + b)); +/// +/// source: +/// 1--2-----4 +/// other: +/// ------3--- +/// result: +/// ------5--7 +/// +/// The result stream will not close until both the source stream and [other] +/// have closed. +/// +/// Errors thrown by [combine], along with any errors on the source stream or +/// [other], are forwarded to the result stream. +/// +/// If the source stream is a broadcast stream, the result stream will be as +/// well, regardless of [other]'s type. If a single subscription stream is +/// combined with a broadcast stream it may never be canceled. +StreamTransformer<S, R> combineLatest<S, T, R>( + Stream<T> other, FutureOr<R> Function(S, T) combine) => + _CombineLatest(other, combine); + +class _CombineLatest<S, T, R> extends StreamTransformerBase<S, R> { + final Stream<T> _other; + final FutureOr<R> Function(S, T) _combine; + + _CombineLatest(this._other, this._combine); + + @override + Stream<R> bind(Stream<S> source) { + final controller = source.isBroadcast + ? StreamController<R>.broadcast(sync: true) + : StreamController<R>(sync: true); + + final other = (source.isBroadcast && !_other.isBroadcast) + ? _other.asBroadcastStream() + : _other; + + StreamSubscription sourceSubscription; + StreamSubscription otherSubscription; + + var sourceDone = false; + var otherDone = false; + + S latestSource; + T latestOther; + + var sourceStarted = false; + var otherStarted = false; + + void emitCombined() { + if (!sourceStarted || !otherStarted) return; + FutureOr<R> result; + try { + result = _combine(latestSource, latestOther); + } catch (e, s) { + controller.addError(e, s); + return; + } + if (result is Future<R>) { + sourceSubscription.pause(); + otherSubscription.pause(); + result + .then(controller.add, onError: controller.addError) + .whenComplete(() { + sourceSubscription.resume(); + otherSubscription.resume(); + }); + } else { + controller.add(result as R); + } + } + + controller.onListen = () { + assert(sourceSubscription == null); + sourceSubscription = source.listen( + (s) { + sourceStarted = true; + latestSource = s; + emitCombined(); + }, + onError: controller.addError, + onDone: () { + sourceDone = true; + if (otherDone) { + controller.close(); + } else if (!sourceStarted) { + // Nothing can ever be emitted + otherSubscription.cancel(); + controller.close(); + } + }); + otherSubscription = other.listen( + (o) { + otherStarted = true; + latestOther = o; + emitCombined(); + }, + onError: controller.addError, + onDone: () { + otherDone = true; + if (sourceDone) { + controller.close(); + } else if (!otherStarted) { + // Nothing can ever be emitted + sourceSubscription.cancel(); + controller.close(); + } + }); + if (!source.isBroadcast) { + controller + ..onPause = () { + sourceSubscription.pause(); + otherSubscription.pause(); + } + ..onResume = () { + sourceSubscription.resume(); + otherSubscription.resume(); + }; + } + controller.onCancel = () { + var cancelSource = sourceSubscription.cancel(); + var cancelOther = otherSubscription.cancel(); + sourceSubscription = null; + otherSubscription = null; + return Future.wait([cancelSource, cancelOther]); + }; + }; + return controller.stream; + } +}
diff --git a/pkgs/stream_transform/lib/stream_transform.dart b/pkgs/stream_transform/lib/stream_transform.dart index bddb744..557fc82 100644 --- a/pkgs/stream_transform/lib/stream_transform.dart +++ b/pkgs/stream_transform/lib/stream_transform.dart
@@ -7,6 +7,7 @@ export 'src/audit.dart'; export 'src/buffer.dart'; export 'src/chain_transformers.dart'; +export 'src/combine_latest.dart'; export 'src/concat.dart'; export 'src/concurrent_async_map.dart'; export 'src/debounce.dart';
diff --git a/pkgs/stream_transform/test/combine_latest_test.dart b/pkgs/stream_transform/test/combine_latest_test.dart new file mode 100644 index 0000000..cbe3050 --- /dev/null +++ b/pkgs/stream_transform/test/combine_latest_test.dart
@@ -0,0 +1,182 @@ +// Copyright (c) 2019, 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:pedantic/pedantic.dart'; +import 'package:test/test.dart'; + +import 'package:stream_transform/stream_transform.dart'; + +void main() { + group('combineLatest', () { + test('flows through combine callback', () async { + var source = StreamController<int>(); + var other = StreamController<int>(); + int sum(int a, int b) => a + b; + + var results = <int>[]; + unawaited(source.stream + .transform(combineLatest(other.stream, sum)) + .forEach(results.add)); + + source.add(1); + await Future(() {}); + expect(results, isEmpty); + + other.add(2); + await Future(() {}); + expect(results, [3]); + + source.add(3); + await Future(() {}); + expect(results, [3, 5]); + + source.add(4); + await Future(() {}); + expect(results, [3, 5, 6]); + + other.add(5); + await Future(() {}); + expect(results, [3, 5, 6, 9]); + }); + + test('can combine different typed streams', () async { + var source = StreamController<String>(); + var other = StreamController<int>(); + String times(String a, int b) => a * b; + + var results = <String>[]; + unawaited(source.stream + .transform(combineLatest(other.stream, times)) + .forEach(results.add)); + + source.add('a'); + source.add('b'); + await Future(() {}); + expect(results, isEmpty); + + other.add(2); + await Future(() {}); + expect(results, ['bb']); + + other.add(3); + await Future(() {}); + expect(results, ['bb', 'bbb']); + + source.add('c'); + await Future(() {}); + expect(results, ['bb', 'bbb', 'ccc']); + }); + + test('ends after both streams have ended', () async { + var source = StreamController<int>(); + var other = StreamController<int>(); + int sum(int a, int b) => a + b; + + var done = false; + source.stream + .transform(combineLatest(other.stream, sum)) + .listen(null, onDone: () => done = true); + + source.add(1); + + await source.close(); + await Future(() {}); + expect(done, false); + + await other.close(); + await Future(() {}); + expect(done, true); + }); + + test('ends if source stream closes without ever emitting a value', + () async { + var source = Stream<int>.empty(); + var other = StreamController<int>(); + + int sum(int a, int b) => a + b; + + var done = false; + source + .transform(combineLatest(other.stream, sum)) + .listen(null, onDone: () => done = true); + + await Future(() {}); + // Nothing can ever be emitted on the result, may as well close. + expect(done, true); + }); + + test('ends if other stream closes without ever emitting a value', () async { + var source = StreamController<int>(); + var other = Stream<int>.empty(); + + int sum(int a, int b) => a + b; + + var done = false; + source.stream + .transform(combineLatest(other, sum)) + .listen(null, onDone: () => done = true); + + await Future(() {}); + // Nothing can ever be emitted on the result, may as well close. + expect(done, true); + }); + + test('forwards errors', () async { + var source = StreamController<int>(); + var other = StreamController<int>(); + int sum(int a, int b) => throw _NumberedException(3); + + var errors = []; + source.stream + .transform(combineLatest(other.stream, sum)) + .listen(null, onError: errors.add); + + source.addError(_NumberedException(1)); + other.addError(_NumberedException(2)); + + source.add(1); + other.add(2); + + await Future(() {}); + + expect(errors, [_isException(1), _isException(2), _isException(3)]); + }); + + group('broadcast source', () { + test('can cancel and relisten to broadcast stream', () async { + var source = StreamController<int>.broadcast(); + var other = StreamController<int>(); + int combine(int a, int b) => a + b; + + var emittedValues = <int>[]; + var transformed = + source.stream.transform(combineLatest(other.stream, combine)); + + var subscription = transformed.listen(emittedValues.add); + + source.add(1); + other.add(2); + await Future(() {}); + expect(emittedValues, [3]); + + await subscription.cancel(); + + subscription = transformed.listen(emittedValues.add); + source.add(3); + await Future(() {}); + expect(emittedValues, [3, 5]); + }); + }); + }); +} + +class _NumberedException implements Exception { + final int id; + _NumberedException(this.id); +} + +Matcher _isException(int id) => + TypeMatcher<_NumberedException>().having((n) => n.id, 'id', id);