Add extension comments and regroup them (dart-lang/stream_transform#84) Add doc comments on all extensions. Merge some related extensions into larger groups and add comments that specifically describe differences between them.
diff --git a/pkgs/stream_transform/README.md b/pkgs/stream_transform/README.md index 313e6f1..785adf1 100644 --- a/pkgs/stream_transform/README.md +++ b/pkgs/stream_transform/README.md
@@ -2,16 +2,17 @@ # Operators -## asyncMapBuffer +## asyncMapBuffer, asyncMapSample, concurrentAsyncMap + +Alternatives to `asyncMap`. `asyncMapBuffer` prevents the callback from +overlapping execution and collects events while it is executing. +`asyncMapSample` prevents overlapping execution and discards events while it is +executing. `concurrentAsyncMap` allows overlap and removes ordering guarantees +for higher throughput. 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. @@ -26,26 +27,16 @@ Collects values from a source stream until a `trigger` stream fires and the collected values are emitted. -## combineLatest +## combineLatest, combineLatestAll -Combine the most recent event from two streams through a callback and emit the -result. - -## combineLatestAll - -Combines the latest events emitted from multiple source streams and yields a -list of the values. +Combine the most recent event from multiple streams through a callback or into a +list. ## debounce, debounceBuffer Prevents a source stream from emitting too frequently by dropping or collecting values that occur within a given duration. -## concurrentAsyncMap - -Like `asyncMap` but the convert callback can be called with subsequent values -before it has finished for previous values. - ## followedBy Appends the values of a stream after another stream finishes.
diff --git a/pkgs/stream_transform/lib/src/async_map_buffer.dart b/pkgs/stream_transform/lib/src/async_map.dart similarity index 63% rename from pkgs/stream_transform/lib/src/async_map_buffer.dart rename to pkgs/stream_transform/lib/src/async_map.dart index 57defc2..c9b263f 100644 --- a/pkgs/stream_transform/lib/src/async_map_buffer.dart +++ b/pkgs/stream_transform/lib/src/async_map.dart
@@ -5,10 +5,20 @@ import 'dart:async'; import 'aggregate_sample.dart'; -import 'buffer.dart'; import 'chain_transformers.dart'; import 'from_handlers.dart'; +import 'rate_limit.dart'; +/// Alternatives to [asyncMap]. +/// +/// The built in [asyncMap] will not overlap execution of the passed callback, +/// and every event will be sent to the callback individually. +/// +/// - [asyncMapBuffer] prevents the callback from overlapping execution and +/// collects events while it is executing to process in batches. +/// - [asyncMapSample] prevents overlapping execution and discards events while +/// it is executing. +/// - [concurrentAsyncMap] allows overlap and removes ordering guarantees. extension AsyncMap<T> on Stream<T> { /// Like [asyncMap] but events are buffered until previous events have been /// processed by [convert]. @@ -63,6 +73,43 @@ return transform(AggregateSample(workFinished.stream, _dropPrevious)) .transform(_asyncMapThen(convert, workFinished.add)); } + + /// Like [asyncMap] but the [convert] callback may be called for an element + /// before processing for the previous element is finished. + /// + /// Events on the result stream will be emitted in the order that [convert] + /// completed which may not match the order of the original stream. + /// + /// If the source stream is a broadcast stream the result will be as well. + /// When used with a broadcast stream behavior also differs from [asyncMap] in + /// that the [convert] function is only called once per event, rather than + /// once per listener per event. The [convert] callback won't be called for + /// events while a broadcast stream has no listener. + /// + /// Errors from [convert] or the source stream are forwarded directly to the + /// result stream. + /// + /// The result stream will not close until the source stream closes and all + /// pending conversions have finished. + Stream<S> concurrentAsyncMap<S>(FutureOr<S> convert(T event)) { + var valuesWaiting = 0; + var sourceDone = false; + return transform(fromHandlers(handleData: (element, sink) { + valuesWaiting++; + () async { + try { + sink.add(await convert(element)); + } catch (e, st) { + sink.addError(e, st); + } + valuesWaiting--; + if (valuesWaiting <= 0 && sourceDone) sink.close(); + }(); + }, handleDone: (sink) { + sourceDone = true; + if (valuesWaiting <= 0) sink.close(); + })); + } } /// Like [Stream.asyncMap] but events are buffered until previous events have @@ -140,3 +187,41 @@ } }); } + +/// Like [Stream.asyncMap] but the [convert] callback may be called for an +/// element before processing for the previous element is finished. +/// +/// Events on the result stream will be emitted in the order that [convert] +/// completed which may not match the order of the original stream. +/// +/// 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. The [convert] callback won't be called for events +/// while a broadcast stream has no listener. +/// +/// Errors from the source stream are forwarded directly to the result stream. +/// Errors during the conversion are also forwarded to the result stream. +/// +/// The result stream will not close until the source stream closes and all +/// pending conversions have finished. +@Deprecated('Use the extension instead') +StreamTransformer<S, T> concurrentAsyncMap<S, T>(FutureOr<T> convert(S event)) { + var valuesWaiting = 0; + var sourceDone = false; + return fromHandlers(handleData: (element, sink) { + valuesWaiting++; + () async { + try { + sink.add(await convert(element)); + } catch (e, st) { + sink.addError(e, st); + } + valuesWaiting--; + if (valuesWaiting <= 0 && sourceDone) sink.close(); + }(); + }, handleDone: (sink) { + sourceDone = true; + if (valuesWaiting <= 0) sink.close(); + }); +}
diff --git a/pkgs/stream_transform/lib/src/async_where.dart b/pkgs/stream_transform/lib/src/async_where.dart index a611063..f2fe108 100644 --- a/pkgs/stream_transform/lib/src/async_where.dart +++ b/pkgs/stream_transform/lib/src/async_where.dart
@@ -5,6 +5,7 @@ import 'from_handlers.dart'; +/// An asynchronous [where]. extension AsyncWhere<T> on Stream<T> { /// Like [where] but allows the [test] to return a [Future]. ///
diff --git a/pkgs/stream_transform/lib/src/audit.dart b/pkgs/stream_transform/lib/src/audit.dart deleted file mode 100644 index b175e6d..0000000 --- a/pkgs/stream_transform/lib/src/audit.dart +++ /dev/null
@@ -1,93 +0,0 @@ -// 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 'from_handlers.dart'; - -extension Audit<T> on Stream<T> { - /// Returns a Stream which only emits once per [duration], at the end of the - /// period. - /// - /// If the source stream is a broadcast stream, the result will be as well. - /// Errors are forwarded immediately. - /// - /// If there is no pending event when the source stream closes the output - /// stream will close immediately. If there is a pending event the output - /// stream will wait to emit it before closing. - /// - /// Differs from `throttle` in that it always emits the most recently received - /// event rather than the first in the period. The events that are emitted are - /// always delayed by some amount. If the event that started the period is the - /// one that is emitted it will be delayed by [duration]. If a later event - /// comes in within the period it's delay will be shorter by the difference in - /// arrival times. - /// - /// Differs from `debounce` in that a value will always be emitted after - /// [duration], the output will not be starved by values coming in repeatedly - /// within [duration]. - /// - /// For example: - /// - /// source.audit(Duration(seconds: 5)); - /// - /// source: a------b--c----d--| - /// output: -----a------c--------d| - Stream<T> audit(Duration duration) { - Timer timer; - var shouldClose = false; - T recentData; - - return transform(fromHandlers(handleData: (T data, EventSink<T> sink) { - recentData = data; - timer ??= Timer(duration, () { - sink.add(recentData); - timer = null; - if (shouldClose) { - sink.close(); - } - }); - }, handleDone: (EventSink<T> sink) { - if (timer != null) { - shouldClose = true; - } else { - sink.close(); - } - })); - } -} - -/// Creates a StreamTransformer which only emits once per [duration], at the -/// end of the period. -/// -/// Always introduces a delay of at most [duration]. -/// -/// Differs from `throttle` in that it always emits the most recently received -/// event rather than the first in the period. -/// -/// Differs from `debounce` in that a value will always be emitted after -/// [duration], the output will not be starved by values coming in repeatedly -/// within [duration]. -@Deprecated('Use the extension instead') -StreamTransformer<T, T> audit<T>(Duration duration) { - Timer timer; - var shouldClose = false; - T recentData; - - return fromHandlers(handleData: (T data, EventSink<T> sink) { - recentData = data; - timer ??= Timer(duration, () { - sink.add(recentData); - timer = null; - if (shouldClose) { - sink.close(); - } - }); - }, handleDone: (EventSink<T> sink) { - if (timer != null) { - shouldClose = true; - } else { - sink.close(); - } - }); -}
diff --git a/pkgs/stream_transform/lib/src/buffer.dart b/pkgs/stream_transform/lib/src/buffer.dart deleted file mode 100644 index cf8b0c3..0000000 --- a/pkgs/stream_transform/lib/src/buffer.dart +++ /dev/null
@@ -1,37 +0,0 @@ -// 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'; - -extension Buffer<T> on Stream<T> { - /// Returns a Stream which collects values and emits when it sees a value on - /// [trigger]. - /// - /// If there are no pending values when [trigger] emits, the next value on the - /// source Stream will immediately flow through. Otherwise, the pending values - /// are released when [trigger] emits. - /// - /// If the source stream is a broadcast stream, the result will be as well. - /// Errors from the source stream or the trigger are immediately forwarded to - /// the output. - Stream<List<T>> buffer(Stream<void> trigger) => - transform(AggregateSample<T, List<T>>(trigger, _collect)); -} - -/// Creates a [StreamTransformer] which collects values and emits when it sees a -/// value on [trigger]. -/// -/// If there are no pending values when [trigger] emits, the next value on the -/// source Stream will immediately flow through. Otherwise, the pending values -/// are released when [trigger] emits. -/// -/// Errors from the source stream or the trigger are immediately forwarded to -/// the output. -@Deprecated('Use the extension instead') -StreamTransformer<T, List<T>> buffer<T>(Stream<void> trigger) => - AggregateSample<T, List<T>>(trigger, _collect); - -List<T> _collect<T>(T event, List<T> soFar) => (soFar ?? <T>[])..add(event);
diff --git a/pkgs/stream_transform/lib/src/combine_latest.dart b/pkgs/stream_transform/lib/src/combine_latest.dart index 7fb9ac9..4e35997 100644 --- a/pkgs/stream_transform/lib/src/combine_latest.dart +++ b/pkgs/stream_transform/lib/src/combine_latest.dart
@@ -4,6 +4,8 @@ import 'dart:async'; +/// Utilities to combine events from multiple streams through a callback or into +/// a list. extension CombineLatest<T> on Stream<T> { /// Returns a stream which combines the latest value from the source stream /// with the latest value from [other] using [combine].
diff --git a/pkgs/stream_transform/lib/src/concat.dart b/pkgs/stream_transform/lib/src/concat.dart deleted file mode 100644 index 565a290..0000000 --- a/pkgs/stream_transform/lib/src/concat.dart +++ /dev/null
@@ -1,9 +0,0 @@ -// Copyright (c) 2018, 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 'followed_by.dart'; - -@Deprecated('Use followedBy instead') -StreamTransformer<T, T> concat<T>(Stream<T> next) => followedBy<T>(next);
diff --git a/pkgs/stream_transform/lib/src/concatenate.dart b/pkgs/stream_transform/lib/src/concatenate.dart new file mode 100644 index 0000000..402a4bc --- /dev/null +++ b/pkgs/stream_transform/lib/src/concatenate.dart
@@ -0,0 +1,169 @@ +// 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'; + +/// Utilities to append or prepend to a stream. +extension Concatenate<T> on Stream<T> { + /// Returns a stream which emits values and errors from [next] after the + /// original stream is complete. + /// + /// If the source stream never finishes, the [next] stream will never be + /// listened to. + /// + /// If the source stream is a broadcast stream, the result will be as well. + /// If a single-subscription follows a broadcast stream it may be listened + /// to and never canceled since there may be broadcast listeners added later. + /// + /// If a broadcast stream follows any other stream it will miss any events or + /// errors which occur before the first stream is done. If a broadcast stream + /// follows a single-subscription stream, pausing the stream while it is + /// listening to the second stream will cause events to be dropped rather than + /// buffered. + Stream<T> followedBy(Stream<T> next) => transform(_FollowedBy(next)); + + /// Returns a stream which emits [initial] before any values from the original + /// stream. + /// + /// If the original stream is a broadcast stream the result will be as well. + Stream<T> startWith(T initial) => + startWithStream(Future.value(initial).asStream()); + + /// Returns a stream which emits all values in [initial] before any values + /// from the original stream. + /// + /// If the original stream is a broadcast stream the result will be as well. + /// If the original stream is a broadcast stream it will miss any events which + /// occur before the initial values are all emitted. + Stream<T> startWithMany(Iterable<T> initial) => + startWithStream(Stream.fromIterable(initial)); + + /// Returns a stream which emits all values in [initial] before any values + /// from the original stream. + /// + /// If the original stream is a broadcast stream the result will be as well. If + /// the original stream is a broadcast stream it will miss any events which + /// occur before [initial] closes. + Stream<T> startWithStream(Stream<T> initial) { + if (isBroadcast && !initial.isBroadcast) { + initial = initial.asBroadcastStream(); + } + return initial.followedBy(this); + } +} + +/// Starts emitting values from [next] after the original stream is complete. +/// +/// If the initial stream never finishes, the [next] stream will never be +/// listened to. +/// +/// If a single-subscription follows the a broadcast stream it may be listened +/// to and never canceled. +/// +/// If a broadcast stream follows any other stream it will miss any events which +/// occur before the first stream is done. If a broadcast stream follows a +/// single-subscription stream, pausing the stream while it is listening to the +/// second stream will cause events to be dropped rather than buffered. +@Deprecated('Use the extension instead') +StreamTransformer<T, T> followedBy<T>(Stream<T> next) => _FollowedBy<T>(next); + +class _FollowedBy<T> extends StreamTransformerBase<T, T> { + final Stream<T> _next; + + _FollowedBy(this._next); + + @override + Stream<T> bind(Stream<T> first) { + var controller = first.isBroadcast + ? StreamController<T>.broadcast(sync: true) + : StreamController<T>(sync: true); + + var next = first.isBroadcast && !_next.isBroadcast + ? _next.asBroadcastStream() + : _next; + + StreamSubscription<T> subscription; + var currentStream = first; + var firstDone = false; + var secondDone = false; + + Function currentDoneHandler; + + listen() { + subscription = currentStream.listen(controller.add, + onError: controller.addError, onDone: () => currentDoneHandler()); + } + + onSecondDone() { + secondDone = true; + controller.close(); + } + + onFirstDone() { + firstDone = true; + currentStream = next; + currentDoneHandler = onSecondDone; + listen(); + } + + currentDoneHandler = onFirstDone; + + controller.onListen = () { + assert(subscription == null); + listen(); + if (!first.isBroadcast) { + controller + ..onPause = () { + if (!firstDone || !next.isBroadcast) return subscription.pause(); + subscription.cancel(); + subscription = null; + } + ..onResume = () { + if (!firstDone || !next.isBroadcast) return subscription.resume(); + listen(); + }; + } + controller.onCancel = () { + if (secondDone) return null; + var toCancel = subscription; + subscription = null; + return toCancel.cancel(); + }; + }; + return controller.stream; + } +} + +/// Emits [initial] before any values from the original stream. +/// +/// If the original stream is a broadcast stream the result will be as well. +@Deprecated('Use the extension instead') +StreamTransformer<T, T> startWith<T>(T initial) => + startWithStream<T>(Future.value(initial).asStream()); + +/// Emits all values in [initial] before any values from the original stream. +/// +/// If the original stream is a broadcast stream the result will be as well. If +/// the original stream is a broadcast stream it will miss any events which +/// occur before the initial values are all emitted. +@Deprecated('Use the extension instead') +StreamTransformer<T, T> startWithMany<T>(Iterable<T> initial) => + startWithStream<T>(Stream.fromIterable(initial)); + +/// Emits all values in [initial] before any values from the original stream. +/// +/// If the original stream is a broadcast stream the result will be as well. If +/// the original stream is a broadcast stream it will miss any events which +/// occur before [initial] closes. +@Deprecated('Use the extension instead') +StreamTransformer<T, T> startWithStream<T>(Stream<T> initial) => + StreamTransformer.fromBind((values) { + if (values.isBroadcast && !initial.isBroadcast) { + initial = initial.asBroadcastStream(); + } + return initial.transform(followedBy(values)); + }); + +@Deprecated('Use followedBy instead') +StreamTransformer<T, T> concat<T>(Stream<T> next) => followedBy<T>(next);
diff --git a/pkgs/stream_transform/lib/src/concurrent_async_map.dart b/pkgs/stream_transform/lib/src/concurrent_async_map.dart deleted file mode 100644 index c0fae2f..0000000 --- a/pkgs/stream_transform/lib/src/concurrent_async_map.dart +++ /dev/null
@@ -1,84 +0,0 @@ -// Copyright (c) 2018, 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 'from_handlers.dart'; - -extension ConcurrentAsyncMap<T> on Stream<T> { - /// Like [asyncMap] but the [convert] callback may be called for an element - /// before processing for the previous element is finished. - /// - /// Events on the result stream will be emitted in the order that [convert] - /// completed which may not match the order of the original stream. - /// - /// If the source stream is a broadcast stream the result will be as well. - /// When used with a broadcast stream behavior also differs from [asyncMap] in - /// that the [convert] function is only called once per event, rather than - /// once per listener per event. The [convert] callback won't be called for - /// events while a broadcast stream has no listener. - /// - /// Errors from [convert] or the source stream are forwarded directly to the - /// result stream. - /// - /// The result stream will not close until the source stream closes and all - /// pending conversions have finished. - Stream<S> concurrentAsyncMap<S>(FutureOr<S> convert(T event)) { - var valuesWaiting = 0; - var sourceDone = false; - return transform(fromHandlers(handleData: (element, sink) { - valuesWaiting++; - () async { - try { - sink.add(await convert(element)); - } catch (e, st) { - sink.addError(e, st); - } - valuesWaiting--; - if (valuesWaiting <= 0 && sourceDone) sink.close(); - }(); - }, handleDone: (sink) { - sourceDone = true; - if (valuesWaiting <= 0) sink.close(); - })); - } -} - -/// Like [Stream.asyncMap] but the [convert] callback may be called for an -/// element before processing for the previous element is finished. -/// -/// Events on the result stream will be emitted in the order that [convert] -/// completed which may not match the order of the original stream. -/// -/// 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. The [convert] callback won't be called for events -/// while a broadcast stream has no listener. -/// -/// Errors from the source stream are forwarded directly to the result stream. -/// Errors during the conversion are also forwarded to the result stream. -/// -/// The result stream will not close until the source stream closes and all -/// pending conversions have finished. -@Deprecated('Use the extension instead') -StreamTransformer<S, T> concurrentAsyncMap<S, T>(FutureOr<T> convert(S event)) { - var valuesWaiting = 0; - var sourceDone = false; - return fromHandlers(handleData: (element, sink) { - valuesWaiting++; - () async { - try { - sink.add(await convert(element)); - } catch (e, st) { - sink.addError(e, st); - } - valuesWaiting--; - if (valuesWaiting <= 0 && sourceDone) sink.close(); - }(); - }, handleDone: (sink) { - sourceDone = true; - if (valuesWaiting <= 0) sink.close(); - }); -}
diff --git a/pkgs/stream_transform/lib/src/debounce.dart b/pkgs/stream_transform/lib/src/debounce.dart deleted file mode 100644 index 0987ec6..0000000 --- a/pkgs/stream_transform/lib/src/debounce.dart +++ /dev/null
@@ -1,100 +0,0 @@ -// 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 'from_handlers.dart'; - -extension Debounce<T> on Stream<T> { - /// Returns a Stream which only emits when the source stream does not emit for - /// [duration]. - /// - /// Values will always be delayed by at least [duration], and values which - /// come within this time will replace the old values, only the most - /// recent value will be emitted. - /// - /// If the source stream is a broadcast stream, the result will be as well. - /// Errors are forwarded immediately. - /// - /// If there is an event waiting during the debounce period when the source - /// stream closes the returned stream will wait to emit it following the - /// debounce period before closing. If there is no pending debounced event - /// when the source stream closes the returned stream will close immediately. - /// - /// To collect values emitted during the debounce period see [debounceBuffer]. - Stream<T> debounce(Duration duration) => - transform(_debounceAggregate(duration, _dropPrevious)); - - /// Returns a Stream which collects values until the source stream does not - /// emit for [duration] then emits the collected values. - /// - /// Values will always be delayed by at least [duration], and values which - /// come within this time will be aggregated into the same list. - /// - /// If the source stream is a broadcast stream, the result will be as well. - /// Errors are forwarded immediately. - /// - /// If there are events waiting during the debounce period when the source - /// stream closes the returned stream will wait to emit them following the - /// debounce period before closing. If there are no pending debounced events - /// when the source stream closes the returned stream will close immediately. - /// - /// To keep only the most recent event during the debounce perios see - /// [debounce]. - Stream<List<T>> debounceBuffer(Duration duration) => - transform(_debounceAggregate(duration, _collectToList)); -} - -/// Creates a StreamTransformer which only emits when the source stream does not -/// emit for [duration]. -/// -/// Source values will always be delayed by at least [duration], and values -/// which come within this time will replace the old values, only the most -/// recent value will be emitted. -@Deprecated('Use the extension instead') -StreamTransformer<T, T> debounce<T>(Duration duration) => - _debounceAggregate(duration, _dropPrevious); - -/// Creates a StreamTransformer which collects values until the source stream -/// does not emit for [duration] then emits the collected values. -/// -/// This differs from [debounce] in that values are aggregated instead of -/// skipped. -@Deprecated('Use the extension instead') -StreamTransformer<T, List<T>> debounceBuffer<T>(Duration duration) => - _debounceAggregate(duration, _collectToList); - -List<T> _collectToList<T>(T element, List<T> soFar) { - soFar ??= <T>[]; - soFar.add(element); - return soFar; -} - -T _dropPrevious<T>(T element, _) => element; - -/// Creates a StreamTransformer which aggregates values until the source stream -/// does not emit for [duration], then emits the aggregated values. -StreamTransformer<T, R> _debounceAggregate<T, R>( - Duration duration, R collect(T element, R soFar)) { - Timer timer; - R soFar; - var shouldClose = false; - return fromHandlers(handleData: (T value, EventSink<R> sink) { - timer?.cancel(); - timer = Timer(duration, () { - sink.add(soFar); - if (shouldClose) { - sink.close(); - } - soFar = null; - timer = null; - }); - soFar = collect(value, soFar); - }, handleDone: (EventSink<R> sink) { - if (soFar != null) { - shouldClose = true; - } else { - sink.close(); - } - }); -}
diff --git a/pkgs/stream_transform/lib/src/followed_by.dart b/pkgs/stream_transform/lib/src/followed_by.dart deleted file mode 100644 index 41c7718..0000000 --- a/pkgs/stream_transform/lib/src/followed_by.dart +++ /dev/null
@@ -1,106 +0,0 @@ -// 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'; - -extension FollowedBy<T> on Stream<T> { - /// Returns a stream which emits values and errors from [next] after the - /// original stream is complete. - /// - /// If the source stream never finishes, the [next] stream will never be - /// listened to. - /// - /// If the source stream is a broadcast stream, the result will be as well. - /// If a single-subscription follows a broadcast stream it may be listened - /// to and never canceled since there may be broadcast listeners added later. - /// - /// If a broadcast stream follows any other stream it will miss any events or - /// errors which occur before the first stream is done. If a broadcast stream - /// follows a single-subscription stream, pausing the stream while it is - /// listening to the second stream will cause events to be dropped rather than - /// buffered. - Stream<T> followedBy(Stream<T> next) => transform(_FollowedBy(next)); -} - -/// Starts emitting values from [next] after the original stream is complete. -/// -/// If the initial stream never finishes, the [next] stream will never be -/// listened to. -/// -/// If a single-subscription follows the a broadcast stream it may be listened -/// to and never canceled. -/// -/// If a broadcast stream follows any other stream it will miss any events which -/// occur before the first stream is done. If a broadcast stream follows a -/// single-subscription stream, pausing the stream while it is listening to the -/// second stream will cause events to be dropped rather than buffered. -@Deprecated('Use the extension instead') -StreamTransformer<T, T> followedBy<T>(Stream<T> next) => _FollowedBy<T>(next); - -class _FollowedBy<T> extends StreamTransformerBase<T, T> { - final Stream<T> _next; - - _FollowedBy(this._next); - - @override - Stream<T> bind(Stream<T> first) { - var controller = first.isBroadcast - ? StreamController<T>.broadcast(sync: true) - : StreamController<T>(sync: true); - - var next = first.isBroadcast && !_next.isBroadcast - ? _next.asBroadcastStream() - : _next; - - StreamSubscription<T> subscription; - var currentStream = first; - var firstDone = false; - var secondDone = false; - - Function currentDoneHandler; - - listen() { - subscription = currentStream.listen(controller.add, - onError: controller.addError, onDone: () => currentDoneHandler()); - } - - onSecondDone() { - secondDone = true; - controller.close(); - } - - onFirstDone() { - firstDone = true; - currentStream = next; - currentDoneHandler = onSecondDone; - listen(); - } - - currentDoneHandler = onFirstDone; - - controller.onListen = () { - assert(subscription == null); - listen(); - if (!first.isBroadcast) { - controller - ..onPause = () { - if (!firstDone || !next.isBroadcast) return subscription.pause(); - subscription.cancel(); - subscription = null; - } - ..onResume = () { - if (!firstDone || !next.isBroadcast) return subscription.resume(); - listen(); - }; - } - controller.onCancel = () { - if (secondDone) return null; - var toCancel = subscription; - subscription = null; - return toCancel.cancel(); - }; - }; - return controller.stream; - } -}
diff --git a/pkgs/stream_transform/lib/src/merge.dart b/pkgs/stream_transform/lib/src/merge.dart index 878a8d1..736cde5 100644 --- a/pkgs/stream_transform/lib/src/merge.dart +++ b/pkgs/stream_transform/lib/src/merge.dart
@@ -4,6 +4,7 @@ import 'dart:async'; +/// Utilities to interleave events from multiple streams. extension Merge<T> on Stream<T> { /// Returns a stream which emits values and errors from the source stream and /// [other] in any order as they arrive.
diff --git a/pkgs/stream_transform/lib/src/rate_limit.dart b/pkgs/stream_transform/lib/src/rate_limit.dart new file mode 100644 index 0000000..91a7648 --- /dev/null +++ b/pkgs/stream_transform/lib/src/rate_limit.dart
@@ -0,0 +1,256 @@ +// 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 'aggregate_sample.dart'; +import 'from_handlers.dart'; + +/// Utilities to rate limit events. +/// +/// - [debounce] - emit the _first_ event at the _end_ of the period. +/// - [debounceBuffer] - emit _all_ events at the _end_ of the period. +/// - [throttle] - emit the _first_ event at the _beginning_ of the period. +/// - [audit] - emit the _last_ event at the _end_ of the period. +/// - [buffer] - emit _all_ events on a _trigger_. +extension RateLimit<T> on Stream<T> { + /// Returns a Stream which only emits when the source stream does not emit for + /// [duration]. + /// + /// Values will always be delayed by at least [duration], and values which + /// come within this time will replace the old values, only the most + /// recent value will be emitted. + /// + /// If the source stream is a broadcast stream, the result will be as well. + /// Errors are forwarded immediately. + /// + /// If there is an event waiting during the debounce period when the source + /// stream closes the returned stream will wait to emit it following the + /// debounce period before closing. If there is no pending debounced event + /// when the source stream closes the returned stream will close immediately. + /// + /// To collect values emitted during the debounce period see [debounceBuffer]. + Stream<T> debounce(Duration duration) => + transform(_debounceAggregate(duration, _dropPrevious)); + + /// Returns a Stream which collects values until the source stream does not + /// emit for [duration] then emits the collected values. + /// + /// Values will always be delayed by at least [duration], and values which + /// come within this time will be aggregated into the same list. + /// + /// If the source stream is a broadcast stream, the result will be as well. + /// Errors are forwarded immediately. + /// + /// If there are events waiting during the debounce period when the source + /// stream closes the returned stream will wait to emit them following the + /// debounce period before closing. If there are no pending debounced events + /// when the source stream closes the returned stream will close immediately. + /// + /// To keep only the most recent event during the debounce perios see + /// [debounce]. + Stream<List<T>> debounceBuffer(Duration duration) => + transform(_debounceAggregate(duration, _collectToList)); + + /// Returns a stream which only emits once per [duration], at the beginning of + /// the period. + /// + /// Events emitted by the source stream within [duration] following an emitted + /// event will be discarded. Errors are always forwarded immediately. + Stream<T> throttle(Duration duration) { + Timer timer; + + return transform(fromHandlers(handleData: (data, sink) { + if (timer == null) { + sink.add(data); + timer = Timer(duration, () { + timer = null; + }); + } + })); + } + + /// Returns a Stream which only emits once per [duration], at the end of the + /// period. + /// + /// If the source stream is a broadcast stream, the result will be as well. + /// Errors are forwarded immediately. + /// + /// If there is no pending event when the source stream closes the output + /// stream will close immediately. If there is a pending event the output + /// stream will wait to emit it before closing. + /// + /// Differs from `throttle` in that it always emits the most recently received + /// event rather than the first in the period. The events that are emitted are + /// always delayed by some amount. If the event that started the period is the + /// one that is emitted it will be delayed by [duration]. If a later event + /// comes in within the period it's delay will be shorter by the difference in + /// arrival times. + /// + /// Differs from `debounce` in that a value will always be emitted after + /// [duration], the output will not be starved by values coming in repeatedly + /// within [duration]. + /// + /// For example: + /// + /// source.audit(Duration(seconds: 5)); + /// + /// source: a------b--c----d--| + /// output: -----a------c--------d| + Stream<T> audit(Duration duration) { + Timer timer; + var shouldClose = false; + T recentData; + + return transform(fromHandlers(handleData: (T data, EventSink<T> sink) { + recentData = data; + timer ??= Timer(duration, () { + sink.add(recentData); + timer = null; + if (shouldClose) { + sink.close(); + } + }); + }, handleDone: (EventSink<T> sink) { + if (timer != null) { + shouldClose = true; + } else { + sink.close(); + } + })); + } + + /// Returns a Stream which collects values and emits when it sees a value on + /// [trigger]. + /// + /// If there are no pending values when [trigger] emits, the next value on the + /// source Stream will immediately flow through. Otherwise, the pending values + /// are released when [trigger] emits. + /// + /// If the source stream is a broadcast stream, the result will be as well. + /// Errors from the source stream or the trigger are immediately forwarded to + /// the output. + Stream<List<T>> buffer(Stream<void> trigger) => + transform(AggregateSample<T, List<T>>(trigger, _collect)); +} + +/// Creates a StreamTransformer which only emits when the source stream does not +/// emit for [duration]. +/// +/// Source values will always be delayed by at least [duration], and values +/// which come within this time will replace the old values, only the most +/// recent value will be emitted. +@Deprecated('Use the extension instead') +StreamTransformer<T, T> debounce<T>(Duration duration) => + _debounceAggregate(duration, _dropPrevious); + +/// Creates a StreamTransformer which collects values until the source stream +/// does not emit for [duration] then emits the collected values. +/// +/// This differs from [debounce] in that values are aggregated instead of +/// skipped. +@Deprecated('Use the extension instead') +StreamTransformer<T, List<T>> debounceBuffer<T>(Duration duration) => + _debounceAggregate(duration, _collectToList); + +List<T> _collectToList<T>(T element, List<T> soFar) { + soFar ??= <T>[]; + soFar.add(element); + return soFar; +} + +T _dropPrevious<T>(T element, _) => element; + +/// Creates a StreamTransformer which aggregates values until the source stream +/// does not emit for [duration], then emits the aggregated values. +StreamTransformer<T, R> _debounceAggregate<T, R>( + Duration duration, R collect(T element, R soFar)) { + Timer timer; + R soFar; + var shouldClose = false; + return fromHandlers(handleData: (T value, EventSink<R> sink) { + timer?.cancel(); + timer = Timer(duration, () { + sink.add(soFar); + if (shouldClose) { + sink.close(); + } + soFar = null; + timer = null; + }); + soFar = collect(value, soFar); + }, handleDone: (EventSink<R> sink) { + if (soFar != null) { + shouldClose = true; + } else { + sink.close(); + } + }); +} + +/// Creates a StreamTransformer which only emits once per [duration], at the +/// beginning of the period. +@Deprecated('Use the extension instead') +StreamTransformer<T, T> throttle<T>(Duration duration) { + Timer timer; + + return fromHandlers(handleData: (data, sink) { + if (timer == null) { + sink.add(data); + timer = Timer(duration, () { + timer = null; + }); + } + }); +} + +/// Creates a StreamTransformer which only emits once per [duration], at the +/// end of the period. +/// +/// Always introduces a delay of at most [duration]. +/// +/// Differs from `throttle` in that it always emits the most recently received +/// event rather than the first in the period. +/// +/// Differs from `debounce` in that a value will always be emitted after +/// [duration], the output will not be starved by values coming in repeatedly +/// within [duration]. +@Deprecated('Use the extension instead') +StreamTransformer<T, T> audit<T>(Duration duration) { + Timer timer; + var shouldClose = false; + T recentData; + + return fromHandlers(handleData: (T data, EventSink<T> sink) { + recentData = data; + timer ??= Timer(duration, () { + sink.add(recentData); + timer = null; + if (shouldClose) { + sink.close(); + } + }); + }, handleDone: (EventSink<T> sink) { + if (timer != null) { + shouldClose = true; + } else { + sink.close(); + } + }); +} + +/// Creates a [StreamTransformer] which collects values and emits when it sees a +/// value on [trigger]. +/// +/// If there are no pending values when [trigger] emits, the next value on the +/// source Stream will immediately flow through. Otherwise, the pending values +/// are released when [trigger] emits. +/// +/// Errors from the source stream or the trigger are immediately forwarded to +/// the output. +@Deprecated('Use the extension instead') +StreamTransformer<T, List<T>> buffer<T>(Stream<void> trigger) => + AggregateSample<T, List<T>>(trigger, _collect); + +List<T> _collect<T>(T event, List<T> soFar) => (soFar ?? <T>[])..add(event);
diff --git a/pkgs/stream_transform/lib/src/scan.dart b/pkgs/stream_transform/lib/src/scan.dart index be9899e..bd2073f 100644 --- a/pkgs/stream_transform/lib/src/scan.dart +++ b/pkgs/stream_transform/lib/src/scan.dart
@@ -4,14 +4,15 @@ import 'dart:async'; +/// A utility similar to [fold] which emits intermediate accumulations. extension Scan<T> on Stream<T> { /// Like [fold], but instead of producing a single value it yields each /// intermediate accumulation. /// /// If [combine] returns a Future it will not be called again for subsequent - /// events from the source until it completes, therefor the combine callback - /// is always called for elements in order, and the result stream always - /// maintains the same order as the original. + /// events from the source until it completes, therefore [combine] is always + /// called for elements in order, and the result stream always maintains the + /// same order as the original. Stream<S> scan<S>( S initialValue, FutureOr<S> combine(S previousValue, T element)) { var accumulated = initialValue;
diff --git a/pkgs/stream_transform/lib/src/start_with.dart b/pkgs/stream_transform/lib/src/start_with.dart deleted file mode 100644 index a9ecaef..0000000 --- a/pkgs/stream_transform/lib/src/start_with.dart +++ /dev/null
@@ -1,68 +0,0 @@ -// 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 'followed_by.dart'; - -extension StartWith<T> on Stream<T> { - /// Returns a stream which emits [initial] before any values from the original - /// stream. - /// - /// If the original stream is a broadcast stream the result will be as well. - Stream<T> startWith(T initial) => - startWithStream(Future.value(initial).asStream()); - - /// Returns a stream which emits all values in [initial] before any values - /// from the original stream. - /// - /// If the original stream is a broadcast stream the result will be as well. - /// If the original stream is a broadcast stream it will miss any events which - /// occur before the initial values are all emitted. - Stream<T> startWithMany(Iterable<T> initial) => - startWithStream(Stream.fromIterable(initial)); - - /// Returns a stream which emits all values in [initial] before any values - /// from the original stream. - /// - /// If the original stream is a broadcast stream the result will be as well. If - /// the original stream is a broadcast stream it will miss any events which - /// occur before [initial] closes. - Stream<T> startWithStream(Stream<T> initial) { - if (isBroadcast && !initial.isBroadcast) { - initial = initial.asBroadcastStream(); - } - return initial.followedBy(this); - } -} - -/// Emits [initial] before any values from the original stream. -/// -/// If the original stream is a broadcast stream the result will be as well. -@Deprecated('Use the extension instead') -StreamTransformer<T, T> startWith<T>(T initial) => - startWithStream<T>(Future.value(initial).asStream()); - -/// Emits all values in [initial] before any values from the original stream. -/// -/// If the original stream is a broadcast stream the result will be as well. If -/// the original stream is a broadcast stream it will miss any events which -/// occur before the initial values are all emitted. -@Deprecated('Use the extension instead') -StreamTransformer<T, T> startWithMany<T>(Iterable<T> initial) => - startWithStream<T>(Stream.fromIterable(initial)); - -/// Emits all values in [initial] before any values from the original stream. -/// -/// If the original stream is a broadcast stream the result will be as well. If -/// the original stream is a broadcast stream it will miss any events which -/// occur before [initial] closes. -@Deprecated('Use the extension instead') -StreamTransformer<T, T> startWithStream<T>(Stream<T> initial) => - StreamTransformer.fromBind((values) { - if (values.isBroadcast && !initial.isBroadcast) { - initial = initial.asBroadcastStream(); - } - return initial.transform(followedBy(values)); - });
diff --git a/pkgs/stream_transform/lib/src/switch.dart b/pkgs/stream_transform/lib/src/switch.dart index e4f4787..eb58cc5 100644 --- a/pkgs/stream_transform/lib/src/switch.dart +++ b/pkgs/stream_transform/lib/src/switch.dart
@@ -4,6 +4,8 @@ import 'dart:async'; +/// A utility to take events from the most recent sub stream returned by a +/// callback. extension Switch<T> on Stream<T> { /// Maps events to a Stream and emits values from the most recently created /// Stream. @@ -18,6 +20,7 @@ } } +/// A utility to take events from the most recent sub stream. extension SwitchLatest<T> on Stream<Stream<T>> { /// Emits values from the most recently emitted Stream. /// @@ -32,11 +35,11 @@ /// Maps events to a Stream and emits values from the most recently created /// Stream. /// -/// When the source emits a value it will be converted to a [Stream] using [map] -/// and the output will switch to emitting events from that result. +/// When the source emits a value it will be converted to a [Stream] using +/// [convert] and the output will switch to emitting events from that result. /// /// If the source stream is a broadcast stream, the result stream will be as -/// well, regardless of the types of the streams produced by [map]. +/// well, regardless of the types of the streams produced by [convert]. @Deprecated('Use the extension instead') StreamTransformer<S, T> switchMap<S, T>(Stream<T> convert(S event)) => StreamTransformer.fromBind(
diff --git a/pkgs/stream_transform/lib/src/take_until.dart b/pkgs/stream_transform/lib/src/take_until.dart index d415c24..862222e 100644 --- a/pkgs/stream_transform/lib/src/take_until.dart +++ b/pkgs/stream_transform/lib/src/take_until.dart
@@ -4,8 +4,9 @@ import 'dart:async'; +/// A utility to end a stream based on an external trigger. extension TakeUntil<T> on Stream<T> { - /// Returns a stram which emits values from the source stream until [trigger] + /// Returns a stream which emits values from the source stream until [trigger] /// fires. /// /// Completing [trigger] differs from canceling a subscription in that values
diff --git a/pkgs/stream_transform/lib/src/tap.dart b/pkgs/stream_transform/lib/src/tap.dart index 61ee198..5d8198f 100644 --- a/pkgs/stream_transform/lib/src/tap.dart +++ b/pkgs/stream_transform/lib/src/tap.dart
@@ -5,6 +5,7 @@ import 'from_handlers.dart'; +/// A utility to chain extra behavior on a stream. extension Tap<T> on Stream<T> { /// Taps into this stream to allow additional handling on a single-subscriber /// stream without first wrapping as a broadcast stream.
diff --git a/pkgs/stream_transform/lib/src/throttle.dart b/pkgs/stream_transform/lib/src/throttle.dart deleted file mode 100644 index bea2980..0000000 --- a/pkgs/stream_transform/lib/src/throttle.dart +++ /dev/null
@@ -1,42 +0,0 @@ -// 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 'from_handlers.dart'; - -extension Throttle<T> on Stream<T> { - /// Returns a stream which only emits once per [duration], at the beginning of - /// the period. - /// - /// Events emitted by the source stream within [duration] following an emitted - /// event will be discarded. Errors are always forwarded immediately. - Stream<T> throttle(Duration duration) { - Timer timer; - - return transform(fromHandlers(handleData: (data, sink) { - if (timer == null) { - sink.add(data); - timer = Timer(duration, () { - timer = null; - }); - } - })); - } -} - -/// Creates a StreamTransformer which only emits once per [duration], at the -/// beginning of the period. -@Deprecated('Use the extension instead') -StreamTransformer<T, T> throttle<T>(Duration duration) { - Timer timer; - - return fromHandlers(handleData: (data, sink) { - if (timer == null) { - sink.add(data); - timer = Timer(duration, () { - timer = null; - }); - } - }); -}
diff --git a/pkgs/stream_transform/lib/src/where_type.dart b/pkgs/stream_transform/lib/src/where_type.dart index eb79bb5..bfd82be 100644 --- a/pkgs/stream_transform/lib/src/where_type.dart +++ b/pkgs/stream_transform/lib/src/where_type.dart
@@ -4,6 +4,7 @@ import 'dart:async'; +/// A utility to filter events by type. extension WhereType<T> on Stream<T> { /// Returns a stream which emits only the events which have type [S]. ///
diff --git a/pkgs/stream_transform/lib/stream_transform.dart b/pkgs/stream_transform/lib/stream_transform.dart index 557fc82..4f7ead8 100644 --- a/pkgs/stream_transform/lib/stream_transform.dart +++ b/pkgs/stream_transform/lib/stream_transform.dart
@@ -2,22 +2,16 @@ // 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. -export 'src/async_map_buffer.dart'; +export 'src/async_map.dart'; export 'src/async_where.dart'; -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'; -export 'src/followed_by.dart'; +export 'src/concatenate.dart'; export 'src/map.dart'; export 'src/merge.dart'; +export 'src/rate_limit.dart'; 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'; export 'src/where_type.dart';