Add takeUntil (dart-lang/stream_transform#38) Closes dart-lang/stream_transform#37
diff --git a/pkgs/stream_transform/CHANGELOG.md b/pkgs/stream_transform/CHANGELOG.md index 7788ddf..9c3e057 100644 --- a/pkgs/stream_transform/CHANGELOG.md +++ b/pkgs/stream_transform/CHANGELOG.md
@@ -1,3 +1,7 @@ +## 0.0.8 + +- Add `takeUntil`. + ## 0.0.7 - Bug Fix: Streams produced with `scan` and `switchMap` now correctly report
diff --git a/pkgs/stream_transform/README.md b/pkgs/stream_transform/README.md index 2602243..261972e 100644 --- a/pkgs/stream_transform/README.md +++ b/pkgs/stream_transform/README.md
@@ -1,5 +1,4 @@ -Contains utility methods to create `StreamTransfomer` instances to manipulate -Streams. +Utility methods to create `StreamTransfomer` instances to manipulate Streams. # asyncWhere @@ -42,6 +41,10 @@ Flatten a Stream of Streams into a Stream which forwards values from the most recent Stream +# takeUntil + +Let values through until a Future fires. + # tap Taps into a single-subscriber stream to react to values as they pass, without
diff --git a/pkgs/stream_transform/lib/src/take_until.dart b/pkgs/stream_transform/lib/src/take_until.dart new file mode 100644 index 0000000..27674c2 --- /dev/null +++ b/pkgs/stream_transform/lib/src/take_until.dart
@@ -0,0 +1,56 @@ +// 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 stream until [trigger] fires. +/// +/// Completing [trigger] differs from canceling a subscription in that values +/// which are emitted before the trigger, but have further asynchronous delays +/// in transformations following the takeUtil, will still go through. Cancelling +/// a subscription immediately stops values. +StreamTransformer<T, T> takeUntil<T>(Future trigger) => new _TakeUntil(trigger); + +class _TakeUntil<T> implements StreamTransformer<T, T> { + final Future _trigger; + + _TakeUntil(this._trigger); + + @override + Stream<T> bind(Stream<T> values) { + var controller = values.isBroadcast + ? new StreamController<T>.broadcast(sync: true) + : new StreamController<T>(sync: true); + + StreamSubscription subscription; + var isDone = false; + _trigger.then((_) { + if (isDone) return; + isDone = true; + subscription?.cancel(); + controller.close(); + }); + + controller.onListen = () { + if (isDone) return; + subscription = values.listen(controller.add, onError: controller.addError, + onDone: () { + if (isDone) return; + isDone = true; + controller.close(); + }); + if (!values.isBroadcast) { + controller.onPause = subscription.pause; + controller.onResume = subscription.resume; + } + controller.onCancel = () { + if (isDone) return null; + var toCancel = subscription; + subscription = null; + return toCancel.cancel(); + }; + }; + return controller.stream; + } +}
diff --git a/pkgs/stream_transform/lib/stream_transform.dart b/pkgs/stream_transform/lib/stream_transform.dart index 4983639..55cefa3 100644 --- a/pkgs/stream_transform/lib/stream_transform.dart +++ b/pkgs/stream_transform/lib/stream_transform.dart
@@ -11,5 +11,6 @@ export 'src/scan.dart'; export 'src/start_with.dart'; export 'src/switch.dart'; +export 'src/take_until.dart'; export 'src/tap.dart'; export 'src/throttle.dart';
diff --git a/pkgs/stream_transform/pubspec.yaml b/pkgs/stream_transform/pubspec.yaml index 879c5ab..1dd2f27 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.7 +version: 0.0.8 environment: sdk: ">=1.22.0 <2.0.0"
diff --git a/pkgs/stream_transform/test/take_until_test.dart b/pkgs/stream_transform/test/take_until_test.dart new file mode 100644 index 0000000..1ed2f79 --- /dev/null +++ b/pkgs/stream_transform/test/take_until_test.dart
@@ -0,0 +1,121 @@ +// 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 'package:test/test.dart'; + +import 'package:stream_transform/stream_transform.dart'; + +void main() { + var streamTypes = { + 'single subscription': () => new StreamController(), + 'broadcast': () => new StreamController.broadcast() + }; + for (var streamType in streamTypes.keys) { + group('takeUntil on Stream type [$streamType]', () { + StreamController values; + List emittedValues; + bool valuesCanceled; + bool isDone; + List errors; + Stream transformed; + StreamSubscription subscription; + Completer closeTrigger; + + setUp(() { + valuesCanceled = false; + values = streamTypes[streamType]() + ..onCancel = () { + valuesCanceled = true; + }; + emittedValues = []; + errors = []; + isDone = false; + closeTrigger = new Completer(); + transformed = values.stream.transform(takeUntil(closeTrigger.future)); + subscription = transformed + .listen(emittedValues.add, onError: errors.add, onDone: () { + isDone = true; + }); + }); + + test('forwards cancellation', () async { + await subscription.cancel(); + expect(valuesCanceled, true); + }); + + test('lets values through before trigger', () async { + values.add(1); + values.add(2); + await new Future(() {}); + expect(emittedValues, [1, 2]); + }); + + test('forwards errors', () async { + values.addError('error'); + await new Future(() {}); + expect(errors, ['error']); + }); + + test('sends done if original strem ends', () async { + await values.close(); + expect(isDone, true); + }); + + test('sends done when trigger fires', () async { + closeTrigger.complete(); + await new Future(() {}); + expect(isDone, true); + }); + + test('cancels value subscription when trigger fires', () async { + closeTrigger.complete(); + await new Future(() {}); + expect(valuesCanceled, true); + }); + + if (streamType == 'broadcast') { + test('multiple listeners all get values', () async { + var otherValues = []; + transformed.listen(otherValues.add); + values.add(1); + values.add(2); + await new Future(() {}); + expect(emittedValues, [1, 2]); + expect(otherValues, [1, 2]); + }); + + test('multiple listeners get done when trigger fires', () async { + var otherDone = false; + transformed.listen(null, onDone: () => otherDone = true); + closeTrigger.complete(); + await new Future(() {}); + expect(otherDone, true); + expect(isDone, true); + }); + + test('multiple listeners get done when values end', () async { + var otherDone = false; + transformed.listen(null, onDone: () => otherDone = true); + await values.close(); + expect(otherDone, true); + expect(isDone, true); + }); + + test('can cancel and relisten before trigger fires', () async { + values.add(1); + await new Future(() {}); + await subscription.cancel(); + values.add(2); + await new Future(() {}); + subscription = transformed.listen(emittedValues.add); + values.add(3); + await new Future(() {}); + expect(emittedValues, [1, 3]); + }); + } + }); + } +}