Add `concurrentAsyncMap` (dart-lang/stream_transform#50)
The `Stream.asyncMap` call is always serial and preserves ordering.
diff --git a/pkgs/stream_transform/CHANGELOG.md b/pkgs/stream_transform/CHANGELOG.md
index 2ad693c..a7c0a7e 100644
--- a/pkgs/stream_transform/CHANGELOG.md
+++ b/pkgs/stream_transform/CHANGELOG.md
@@ -2,6 +2,7 @@
- `asyncWhere` will now forward exceptions thrown by the callback through the
result Stream.
+- Added `concurrentAsyncMap`.
## 0.0.13
diff --git a/pkgs/stream_transform/README.md b/pkgs/stream_transform/README.md
index 5e59fc9..5e54457 100644
--- a/pkgs/stream_transform/README.md
+++ b/pkgs/stream_transform/README.md
@@ -2,8 +2,8 @@
# asyncMapBuffer
-Like `asyncMap` but events are buffered until previous events have been
-processed.
+Like `asyncMap` but events are buffered in a List until previous events have
+been processed rather than being called for each element individually.
# asyncWhere
@@ -24,6 +24,11 @@
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/concurrent_async_map.dart b/pkgs/stream_transform/lib/src/concurrent_async_map.dart
new file mode 100644
index 0000000..791c84c
--- /dev/null
+++ b/pkgs/stream_transform/lib/src/concurrent_async_map.dart
@@ -0,0 +1,44 @@
+// 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';
+
+/// 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.
+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/stream_transform.dart b/pkgs/stream_transform/lib/stream_transform.dart
index 942391b..13b80a4 100644
--- a/pkgs/stream_transform/lib/stream_transform.dart
+++ b/pkgs/stream_transform/lib/stream_transform.dart
@@ -8,6 +8,7 @@
export 'src/buffer.dart';
export 'src/chain_transformers.dart';
export 'src/concat.dart';
+export 'src/concurrent_async_map.dart';
export 'src/debounce.dart';
export 'src/followed_by.dart';
export 'src/map.dart';
diff --git a/pkgs/stream_transform/pubspec.yaml b/pkgs/stream_transform/pubspec.yaml
index 53f7bbf..6d1d537 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.13
+version: 0.0.14
environment:
sdk: ">=2.0.0-dev.20.0 <2.0.0"
diff --git a/pkgs/stream_transform/test/concurrent_async_map_test.dart b/pkgs/stream_transform/test/concurrent_async_map_test.dart
new file mode 100644
index 0000000..9ea11a5
--- /dev/null
+++ b/pkgs/stream_transform/test/concurrent_async_map_test.dart
@@ -0,0 +1,160 @@
+// 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 'package:test/test.dart';
+
+import 'package:stream_transform/stream_transform.dart';
+
+void main() {
+ var streamTypes = {
+ 'single subscription': () => new StreamController(),
+ 'broadcast': () => new StreamController.broadcast()
+ };
+ StreamController streamController;
+ List emittedValues;
+ bool valuesCanceled;
+ bool isDone;
+ List errors;
+ Stream transformed;
+ StreamSubscription subscription;
+
+ List<Completer> finishWork;
+ List<dynamic> values;
+
+ Future convert(dynamic value) {
+ values.add(value);
+ var completer = new Completer();
+ finishWork.add(completer);
+ return completer.future;
+ }
+
+ for (var streamType in streamTypes.keys) {
+ group('concurrentAsyncMap for stream type: [$streamType]', () {
+ setUp(() {
+ valuesCanceled = false;
+ streamController = streamTypes[streamType]()
+ ..onCancel = () {
+ valuesCanceled = true;
+ };
+ emittedValues = [];
+ errors = [];
+ isDone = false;
+ finishWork = [];
+ values = [];
+ transformed =
+ streamController.stream.transform(concurrentAsyncMap(convert));
+ subscription = transformed
+ .listen(emittedValues.add, onError: errors.add, onDone: () {
+ isDone = true;
+ });
+ });
+
+ test('does not emit before convert finishes', () async {
+ streamController.add(1);
+ await new Future(() {});
+ expect(emittedValues, isEmpty);
+ expect(values, [1]);
+ finishWork.first.complete(1);
+ await new Future(() {});
+ expect(emittedValues, [1]);
+ });
+
+ test('allows calls to convert before the last one finished', () async {
+ streamController.add(1);
+ streamController.add(2);
+ streamController.add(3);
+ await new Future(() {});
+ expect(values, [1, 2, 3]);
+ });
+
+ test('forwards errors directly without waiting for previous convert',
+ () async {
+ streamController.add(1);
+ await new Future(() {});
+ streamController.addError('error');
+ await new Future(() {});
+ expect(errors, ['error']);
+ });
+
+ test('forwards errors which occur during the convert', () async {
+ streamController.add(1);
+ await new Future(() {});
+ finishWork.first.completeError('error');
+ await new Future(() {});
+ expect(errors, ['error']);
+ });
+
+ test('can continue handling events after an error', () async {
+ streamController.add(1);
+ await new Future(() {});
+ finishWork[0].completeError('error');
+ streamController.add(2);
+ await new Future(() {});
+ expect(values, [1, 2]);
+ finishWork[1].completeError('another');
+ await new Future(() {});
+ expect(errors, ['error', 'another']);
+ });
+
+ test('cancels value subscription when output canceled', () async {
+ expect(valuesCanceled, false);
+ await subscription.cancel();
+ expect(valuesCanceled, true);
+ });
+
+ test('closes when values end if no conversion is pending', () async {
+ expect(isDone, false);
+ await streamController.close();
+ await new Future(() {});
+ expect(isDone, true);
+ });
+
+ if (streamType == 'broadcast') {
+ test('multiple listeners all get values', () async {
+ var otherValues = [];
+ transformed.listen(otherValues.add);
+ streamController.add(1);
+ await new Future(() {});
+ finishWork.first.complete('result');
+ await new 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);
+ streamController.add(1);
+ await new Future(() {});
+ await streamController.close();
+ expect(isDone, false);
+ expect(otherDone, false);
+ finishWork.first.complete();
+ await new Future(() {});
+ expect(isDone, true);
+ expect(otherDone, true);
+ });
+
+ test('can cancel and relisten', () async {
+ streamController.add(1);
+ await new Future(() {});
+ finishWork.first.complete('first');
+ await new Future(() {});
+ await subscription.cancel();
+ streamController.add(2);
+ await new Future(() {});
+ subscription = transformed.listen(emittedValues.add);
+ streamController.add(3);
+ await new Future(() {});
+ expect(values, [1, 3]);
+ finishWork[1].complete('second');
+ await new Future(() {});
+ expect(emittedValues, ['first', 'second']);
+ });
+ }
+ });
+ }
+}
