Add asyncMapSample transform (dart-lang/stream_transform#79) Like `asyncMapBuffer` but drops instead of collecting previous events. - Refactor `_Buffer` class into `Aggregate` which takes a function to aggregate values instead of only collecting them to a List. This will not be published. - Add `asyncMapSample` which is effectively an identical implementation to `asyncMapBuffer` except it chooses a different aggregation function.
diff --git a/pkgs/stream_transform/CHANGELOG.md b/pkgs/stream_transform/CHANGELOG.md index 3f502d3..e631a97 100644 --- a/pkgs/stream_transform/CHANGELOG.md +++ b/pkgs/stream_transform/CHANGELOG.md
@@ -1,3 +1,7 @@ +## 0.0.19 + +- Add `asyncMapSample` transform. + ## 0.0.18 - Internal cleanup. Passed "trigger" streams or futures now allow `<void>`
diff --git a/pkgs/stream_transform/README.md b/pkgs/stream_transform/README.md index 5d24caf..fc44207 100644 --- a/pkgs/stream_transform/README.md +++ b/pkgs/stream_transform/README.md
@@ -5,6 +5,11 @@ Like `asyncMap` but events are buffered in a List until previous events have been processed rather than being called for each element individually. +# asyncMapSample + +Like `asyncMap` but events are discarded, keeping only the latest, until +previous events have been processed rather than being called for every element. + # asyncWhere Like `where` but allows an asynchronous predicate.
diff --git a/pkgs/stream_transform/lib/src/aggregate_sample.dart b/pkgs/stream_transform/lib/src/aggregate_sample.dart new file mode 100644 index 0000000..a069af7 --- /dev/null +++ b/pkgs/stream_transform/lib/src/aggregate_sample.dart
@@ -0,0 +1,116 @@ +// 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'; + +/// A StreamTransformer which aggregates values and emits when it sees a value +/// on [_trigger]. +/// +/// If there are no pending values when [_trigger] emits the first value on the +/// source Stream will immediately flow through. Otherwise, the pending values +/// and released when [_trigger] emits. +/// +/// Errors from the source stream or the trigger are immediately forwarded to +/// the output. +class AggregateSample<S, T> extends StreamTransformerBase<S, T> { + final Stream<void> _trigger; + final T Function(S, T) _aggregate; + + AggregateSample(this._trigger, this._aggregate); + + @override + Stream<T> bind(Stream<S> values) { + var controller = values.isBroadcast + ? StreamController<T>.broadcast(sync: true) + : StreamController<T>(sync: true); + + T currentResults; + var waitingForTrigger = true; + var isTriggerDone = false; + var isValueDone = false; + StreamSubscription<S> valueSub; + StreamSubscription<void> triggerSub; + + emit() { + controller.add(currentResults); + currentResults = null; + waitingForTrigger = true; + } + + onValue(S value) { + currentResults = _aggregate(value, currentResults); + + if (!waitingForTrigger) emit(); + + if (isTriggerDone) { + valueSub.cancel(); + controller.close(); + } + } + + onValuesDone() { + isValueDone = true; + if (currentResults == null) { + triggerSub?.cancel(); + controller.close(); + } + } + + onTrigger(_) { + waitingForTrigger = false; + + if (currentResults != null) emit(); + + if (isValueDone) { + triggerSub.cancel(); + controller.close(); + } + } + + onTriggerDone() { + isTriggerDone = true; + if (waitingForTrigger) { + valueSub?.cancel(); + controller.close(); + } + } + + controller.onListen = () { + assert(valueSub == null); + valueSub = values.listen(onValue, + onError: controller.addError, onDone: onValuesDone); + if (triggerSub != null) { + if (triggerSub.isPaused) triggerSub.resume(); + } else { + triggerSub = _trigger.listen(onTrigger, + onError: controller.addError, onDone: onTriggerDone); + } + if (!values.isBroadcast) { + controller + ..onPause = () { + valueSub?.pause(); + triggerSub?.pause(); + } + ..onResume = () { + valueSub?.resume(); + triggerSub?.resume(); + }; + } + controller.onCancel = () { + var toCancel = <StreamSubscription<void>>[]; + if (!isValueDone) toCancel.add(valueSub); + valueSub = null; + if (_trigger.isBroadcast || !values.isBroadcast) { + if (!isTriggerDone) toCancel.add(triggerSub); + triggerSub = null; + } else { + triggerSub.pause(); + } + if (toCancel.isEmpty) return null; + return Future.wait(toCancel.map((s) => s.cancel())); + }; + }; + return controller.stream; + } +}
diff --git a/pkgs/stream_transform/lib/src/async_map_buffer.dart b/pkgs/stream_transform/lib/src/async_map_buffer.dart index a4e346f..70c5bdb 100644 --- a/pkgs/stream_transform/lib/src/async_map_buffer.dart +++ b/pkgs/stream_transform/lib/src/async_map_buffer.dart
@@ -4,6 +4,7 @@ import 'dart:async'; +import 'aggregate_sample.dart'; import 'buffer.dart'; import 'chain_transformers.dart'; import 'from_handlers.dart'; @@ -27,7 +28,7 @@ /// The result stream will not close until the source stream closes and all /// pending conversions have finished. StreamTransformer<S, T> asyncMapBuffer<S, T>( - Future<T> convert(List<S> collected)) { + Future<T> Function(List<S>) convert) { var workFinished = StreamController<void>() // Let the first event through. ..add(null); @@ -35,6 +36,35 @@ buffer(workFinished.stream), _asyncMapThen(convert, workFinished.add)); } +/// Like [Stream.asyncMap] but events are discarded while work is happening in +/// [convert]. +/// +/// If the source stream is a broadcast stream the result will be as well. When +/// used with a broadcast stream behavior also differs from [Stream.asyncMap] in +/// that the [convert] function is only called once per event, rather than once +/// per listener per event. +/// +/// If no work is happening when an event is emitted it will be immediately +/// passed to [convert]. If there is ongoing work when an event is emitted it +/// will be held until the work is finished. New events emitted will replace a +/// pending event. +/// +/// Errors from the source stream are forwarded directly to the result stream. +/// Errors during the conversion are also forwarded to the result stream and are +/// considered completing work so the next values are let through. +/// +/// The result stream will not close until the source stream closes and all +/// pending conversions have finished. +StreamTransformer<S, T> asyncMapSample<S, T>(Future<T> Function(S) convert) { + var workFinished = StreamController<void>() + // Let the first event through. + ..add(null); + return chainTransformers(AggregateSample(workFinished.stream, _dropPrevious), + _asyncMapThen(convert, workFinished.add)); +} + +T _dropPrevious<T>(T event, _) => event; + /// Like [Stream.asyncMap] but the [convert] is only called once per event, /// rather than once per listener, and [then] is called after completing the /// work.
diff --git a/pkgs/stream_transform/lib/src/buffer.dart b/pkgs/stream_transform/lib/src/buffer.dart index 8641b15..3478a52 100644 --- a/pkgs/stream_transform/lib/src/buffer.dart +++ b/pkgs/stream_transform/lib/src/buffer.dart
@@ -1,8 +1,11 @@ // 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'; +import 'aggregate_sample.dart'; + /// Creates a [StreamTransformer] which collects values and emits when it sees a /// value on [trigger]. /// @@ -13,114 +16,6 @@ /// Errors from the source stream or the trigger are immediately forwarded to /// the output. StreamTransformer<T, List<T>> buffer<T>(Stream<void> trigger) => - _Buffer<T>(trigger); + AggregateSample<T, List<T>>(trigger, _collect); -/// A StreamTransformer which aggregates values and emits when it sees a value -/// on [_trigger]. -/// -/// If there are no pending values when [_trigger] emits the first value on the -/// source Stream will immediately flow through. Otherwise, the pending values -/// and released when [_trigger] emits. -/// -/// Errors from the source stream or the trigger are immediately forwarded to -/// the output. -class _Buffer<T> extends StreamTransformerBase<T, List<T>> { - final Stream<void> _trigger; - - _Buffer(this._trigger); - - @override - Stream<List<T>> bind(Stream<T> values) { - var controller = values.isBroadcast - ? StreamController<List<T>>.broadcast(sync: true) - : StreamController<List<T>>(sync: true); - - List<T> currentResults; - var waitingForTrigger = true; - var isTriggerDone = false; - var isValueDone = false; - StreamSubscription<T> valueSub; - StreamSubscription<void> triggerSub; - - emit() { - controller.add(currentResults); - currentResults = null; - waitingForTrigger = true; - } - - onValue(T value) { - (currentResults ??= <T>[]).add(value); - - if (!waitingForTrigger) emit(); - - if (isTriggerDone) { - valueSub.cancel(); - controller.close(); - } - } - - onValuesDone() { - isValueDone = true; - if (currentResults == null) { - triggerSub?.cancel(); - controller.close(); - } - } - - onTrigger(_) { - waitingForTrigger = false; - - if (currentResults != null) emit(); - - if (isValueDone) { - triggerSub.cancel(); - controller.close(); - } - } - - onTriggerDone() { - isTriggerDone = true; - if (waitingForTrigger) { - valueSub?.cancel(); - controller.close(); - } - } - - controller.onListen = () { - assert(valueSub == null); - valueSub = values.listen(onValue, - onError: controller.addError, onDone: onValuesDone); - if (triggerSub != null) { - if (triggerSub.isPaused) triggerSub.resume(); - } else { - triggerSub = _trigger.listen(onTrigger, - onError: controller.addError, onDone: onTriggerDone); - } - if (!values.isBroadcast) { - controller - ..onPause = () { - valueSub?.pause(); - triggerSub?.pause(); - } - ..onResume = () { - valueSub?.resume(); - triggerSub?.resume(); - }; - } - controller.onCancel = () { - var toCancel = <StreamSubscription<void>>[]; - if (!isValueDone) toCancel.add(valueSub); - valueSub = null; - if (_trigger.isBroadcast || !values.isBroadcast) { - if (!isTriggerDone) toCancel.add(triggerSub); - triggerSub = null; - } else { - triggerSub.pause(); - } - if (toCancel.isEmpty) return null; - return Future.wait(toCancel.map((s) => s.cancel())); - }; - }; - return controller.stream; - } -} +List<T> _collect<T>(T event, List<T> soFar) => (soFar ?? <T>[])..add(event);
diff --git a/pkgs/stream_transform/pubspec.yaml b/pkgs/stream_transform/pubspec.yaml index 37a6cbf..38ad8d5 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.18 +version: 0.0.19 environment: sdk: ">=2.2.0 <3.0.0"
diff --git a/pkgs/stream_transform/test/async_map_sample_test.dart b/pkgs/stream_transform/test/async_map_sample_test.dart new file mode 100644 index 0000000..1a6440e --- /dev/null +++ b/pkgs/stream_transform/test/async_map_sample_test.dart
@@ -0,0 +1,203 @@ +// 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:test/test.dart'; + +import 'package:stream_transform/stream_transform.dart'; + +import 'utils.dart'; + +void main() { + StreamController<int> values; + List<String> emittedValues; + bool valuesCanceled; + bool isDone; + List<String> errors; + Stream<String> transformed; + StreamSubscription<String> subscription; + + Completer<String> finishWork; + int workArgument; + + /// Represents the async `convert` function and asserts that is is only called + /// after the previous iteration has completed. + Future<String> work(int value) { + expect(finishWork, isNull, + reason: 'See $values befor previous work is complete'); + workArgument = value; + finishWork = Completer(); + finishWork.future.then((_) { + workArgument = null; + finishWork = null; + }).catchError((_) { + workArgument = null; + finishWork = null; + }); + return finishWork.future; + } + + for (var streamType in streamTypes) { + group('asyncMapSample for stream type: [$streamType]', () { + setUp(() { + valuesCanceled = false; + values = createController(streamType) + ..onCancel = () { + valuesCanceled = true; + }; + emittedValues = []; + errors = []; + isDone = false; + finishWork = null; + workArgument = null; + transformed = values.stream.transform(asyncMapSample(work)); + subscription = transformed + .listen(emittedValues.add, onError: errors.add, onDone: () { + isDone = true; + }); + }); + + test('does not emit before work finishes', () async { + values.add(1); + await Future(() {}); + expect(emittedValues, isEmpty); + expect(workArgument, 1); + finishWork.complete('result'); + await Future(() {}); + expect(emittedValues, ['result']); + }); + + test('buffers values while work is ongoing', () async { + values.add(1); + await Future(() {}); + values..add(2)..add(3); + await Future(() {}); + finishWork.complete(); + await Future(() {}); + expect(workArgument, 3); + }); + + test('forwards errors without waiting for work', () async { + values.add(1); + await Future(() {}); + values.addError('error'); + await Future(() {}); + expect(errors, ['error']); + }); + + test('forwards errors which occur during the work', () async { + values.add(1); + await Future(() {}); + finishWork.completeError('error'); + await Future(() {}); + expect(errors, ['error']); + }); + + test('can continue handling events after an error', () async { + values.add(1); + await Future(() {}); + finishWork.completeError('error'); + values.add(2); + await Future(() {}); + expect(workArgument, 2); + finishWork.completeError('another'); + await Future(() {}); + expect(errors, ['error', 'another']); + }); + + test('does not start next work early due to an error in values', + () async { + values.add(1); + await Future(() {}); + values + ..addError('error') + ..add(2); + await Future(() {}); + expect(errors, ['error']); + // [work] will assert that the second iteration is not called because + // the first has not completed. + }); + + test('cancels value subscription when output canceled', () async { + expect(valuesCanceled, false); + await subscription.cancel(); + expect(valuesCanceled, true); + }); + + test('closes when values end if no work is pending', () async { + expect(isDone, false); + await values.close(); + await Future(() {}); + expect(isDone, true); + }); + + test('waits for pending work when values close', () async { + values.add(1); + await Future(() {}); + expect(isDone, false); + values.add(2); + await values.close(); + expect(isDone, false); + finishWork.complete(null); + await Future(() {}); + // Still a pending value + expect(isDone, false); + finishWork.complete(null); + await Future(() {}); + expect(isDone, true); + }); + + test('forwards errors from values', () async { + values.addError('error'); + await Future(() {}); + expect(errors, ['error']); + }); + + if (streamType == 'broadcast') { + test('multiple listeners all get values', () async { + var otherValues = []; + transformed.listen(otherValues.add); + values.add(1); + await Future(() {}); + finishWork.complete('result'); + await Future(() {}); + expect(emittedValues, ['result']); + expect(otherValues, ['result']); + }); + + test('multiple listeners get done when values end', () async { + var otherDone = false; + transformed.listen(null, onDone: () => otherDone = true); + values.add(1); + await Future(() {}); + await values.close(); + expect(isDone, false); + expect(otherDone, false); + finishWork.complete(); + await Future(() {}); + expect(isDone, true); + expect(otherDone, true); + }); + + test('can cancel and relisten', () async { + values.add(1); + await Future(() {}); + finishWork.complete('first'); + await Future(() {}); + await subscription.cancel(); + values.add(2); + await Future(() {}); + subscription = transformed.listen(emittedValues.add); + values.add(3); + await Future(() {}); + expect(workArgument, 3); + finishWork.complete('second'); + await Future(() {}); + expect(emittedValues, ['first', 'second']); + }); + } + }); + } +}