Make a separate library for async expand methods (dart-lang/stream_transform#144)
Pull `concurrentAsyncExpand` out from the merge extensions into a new
extension. We will add `asyncExpandSample` in a followup, so having
these in a dedicated library will make it easier for editors to find.
This could be breaking, but only for code that references the extension
name directly, which is an unlikely pattern. No usages of that pattern
exist in Dart code on Github.
Inline the implementation of `concurrentAsyncExpand` instead of
implementing a separate extension on `Stream<Stream>`
diff --git a/pkgs/stream_transform/lib/src/async_expand.dart b/pkgs/stream_transform/lib/src/async_expand.dart
new file mode 100644
index 0000000..265f481
--- /dev/null
+++ b/pkgs/stream_transform/lib/src/async_expand.dart
@@ -0,0 +1,90 @@
+// Copyright (c) 2022, 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 'switch.dart';
+
+/// Alternatives to [asyncExpand].
+///
+/// The built in [asyncExpand] will not overlap the inner streams and every
+/// event will be sent to the callback individually.
+///
+/// - [concurrentAsyncExpand] allow overlap and merges inner streams without
+/// ordering guarantees.
+extension AsyncExpand<T> on Stream<T> {
+ /// Like [asyncExpand] but the [convert] callback may be called for an element
+ /// before the [Stream] emitted by the previous element has closed.
+ ///
+ /// Events on the result stream will be emitted in the order they are emitted
+ /// by the sub streams, which may not match the order of the original stream.
+ ///
+ /// Errors from [convert], the source stream, or any of the sub streams are
+ /// forwarded to the result stream.
+ ///
+ /// The result stream will not close until the source stream closes and all
+ /// sub streams have closed.
+ ///
+ /// If the source stream is a broadcast stream, the result will be as well,
+ /// regardless of the types of streams created by [convert]. In this case,
+ /// some care should be taken:
+ /// - If [convert] returns a single subscription stream it may be listened to
+ /// and never canceled.
+ /// - For any period of time where there are no listeners on the result
+ /// stream, any sub streams from previously emitted events will be ignored,
+ /// regardless of whether they emit further events after a listener is added
+ /// back.
+ ///
+ /// See also:
+ ///
+ /// * [switchMap], which cancels subscriptions to the previous sub
+ /// stream instead of concurrently emitting events from all sub streams.
+ Stream<S> concurrentAsyncExpand<S>(Stream<S> Function(T) convert) {
+ final controller = isBroadcast
+ ? StreamController<S>.broadcast(sync: true)
+ : StreamController<S>(sync: true);
+
+ controller.onListen = () {
+ final subscriptions = <StreamSubscription<dynamic>>[];
+ final outerSubscription = map(convert).listen((inner) {
+ if (isBroadcast && !inner.isBroadcast) {
+ inner = inner.asBroadcastStream();
+ }
+ final subscription =
+ inner.listen(controller.add, onError: controller.addError);
+ subscription.onDone(() {
+ subscriptions.remove(subscription);
+ if (subscriptions.isEmpty) controller.close();
+ });
+ subscriptions.add(subscription);
+ }, onError: controller.addError);
+ outerSubscription.onDone(() {
+ subscriptions.remove(outerSubscription);
+ if (subscriptions.isEmpty) controller.close();
+ });
+ subscriptions.add(outerSubscription);
+ if (!isBroadcast) {
+ controller
+ ..onPause = () {
+ for (final subscription in subscriptions) {
+ subscription.pause();
+ }
+ }
+ ..onResume = () {
+ for (final subscription in subscriptions) {
+ subscription.resume();
+ }
+ };
+ }
+ controller.onCancel = () {
+ if (subscriptions.isEmpty) return null;
+ var cancels = [for (var s in subscriptions) s.cancel()]
+ // Handle opt-out nulls
+ ..removeWhere((Object? f) => f == null);
+ return Future.wait(cancels).then((_) => null);
+ };
+ };
+ return controller.stream;
+ }
+}
diff --git a/pkgs/stream_transform/lib/src/merge.dart b/pkgs/stream_transform/lib/src/merge.dart
index 55fe409..7784bdb 100644
--- a/pkgs/stream_transform/lib/src/merge.dart
+++ b/pkgs/stream_transform/lib/src/merge.dart
@@ -4,8 +4,6 @@
import 'dart:async';
-import 'package:stream_transform/src/switch.dart';
-
/// 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
@@ -101,83 +99,4 @@
};
return controller.stream;
}
-
- /// Like [asyncExpand] but the [convert] callback may be called for an element
- /// before the [Stream] emitted by the previous element has closed.
- ///
- /// Events on the result stream will be emitted in the order they are emitted
- /// by the sub streams, which may not match the order of the original stream.
- ///
- /// Errors from [convert], the source stream, or any of the sub streams are
- /// forwarded to the result stream.
- ///
- /// The result stream will not close until the source stream closes and all
- /// sub streams have closed.
- ///
- /// If the source stream is a broadcast stream, the result will be as well,
- /// regardless of the types of streams created by [convert]. In this case,
- /// some care should be taken:
- /// - If [convert] returns a single subscription stream it may be listened to
- /// and never canceled.
- /// - For any period of time where there are no listeners on the result
- /// stream, any sub streams from previously emitted events will be ignored,
- /// regardless of whether they emit further events after a listener is added
- /// back.
- ///
- /// See also:
- ///
- /// * [switchMap], which cancels subscriptions to the previous sub
- /// stream instead of concurrently emitting events from all sub streams.
- Stream<S> concurrentAsyncExpand<S>(Stream<S> Function(T) convert) =>
- map(convert).mergeExpanded();
-}
-
-extension _MergeExpanded<T> on Stream<Stream<T>> {
- Stream<T> mergeExpanded() {
- final controller = isBroadcast
- ? StreamController<T>.broadcast(sync: true)
- : StreamController<T>(sync: true);
-
- controller.onListen = () {
- final subscriptions = <StreamSubscription<dynamic>>[];
- final outerSubscription = listen((inner) {
- if (isBroadcast && !inner.isBroadcast) {
- inner = inner.asBroadcastStream();
- }
- final subscription =
- inner.listen(controller.add, onError: controller.addError);
- subscription.onDone(() {
- subscriptions.remove(subscription);
- if (subscriptions.isEmpty) controller.close();
- });
- subscriptions.add(subscription);
- }, onError: controller.addError);
- outerSubscription.onDone(() {
- subscriptions.remove(outerSubscription);
- if (subscriptions.isEmpty) controller.close();
- });
- subscriptions.add(outerSubscription);
- if (!isBroadcast) {
- controller
- ..onPause = () {
- for (final subscription in subscriptions) {
- subscription.pause();
- }
- }
- ..onResume = () {
- for (final subscription in subscriptions) {
- subscription.resume();
- }
- };
- }
- controller.onCancel = () {
- if (subscriptions.isEmpty) return null;
- var cancels = [for (var s in subscriptions) s.cancel()]
- // Handle opt-out nulls
- ..removeWhere((Object? f) => f == null);
- return Future.wait(cancels).then((_) => null);
- };
- };
- return controller.stream;
- }
}
diff --git a/pkgs/stream_transform/lib/src/switch.dart b/pkgs/stream_transform/lib/src/switch.dart
index 1fa07da..c749fbf 100644
--- a/pkgs/stream_transform/lib/src/switch.dart
+++ b/pkgs/stream_transform/lib/src/switch.dart
@@ -4,7 +4,7 @@
import 'dart:async';
-import 'package:stream_transform/src/merge.dart';
+import 'async_expand.dart';
/// A utility to take events from the most recent sub stream returned by a
/// callback.
diff --git a/pkgs/stream_transform/lib/stream_transform.dart b/pkgs/stream_transform/lib/stream_transform.dart
index d04c4dd..edf4df9 100644
--- a/pkgs/stream_transform/lib/stream_transform.dart
+++ b/pkgs/stream_transform/lib/stream_transform.dart
@@ -2,6 +2,7 @@
// 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_expand.dart';
export 'src/async_map.dart';
export 'src/combine_latest.dart';
export 'src/concatenate.dart';
diff --git a/pkgs/stream_transform/test/concurrent_async_expand_test.dart b/pkgs/stream_transform/test/async_expand_test.dart
similarity index 100%
rename from pkgs/stream_transform/test/concurrent_async_expand_test.dart
rename to pkgs/stream_transform/test/async_expand_test.dart
