Synchronously forward chained cancellations (#239)
Add a test which chains through multiple `.then` calls and then cancels
the original operation. This test passes with the original code, but
starts failing after fixing a bug in `Completer.complete` in the SDK.
https://dart-review.googlesource.com/c/sdk/+/258929
- Add a list of extra completers to forward cancellation. In places
where the cancellation had been forwarded through a `whenComplete`,
add to the list of extra cancellations instead. This allows the Future
return by the outer cancellation to wait for the other callbacks to
complete before firing.
- When setting up forwarding on an already canceled completer,
immediately forward the cancellation.
- When forwarding cancellations, don't repeatedly call `_cancel()` on a
completer which has already been canceled.
- In `race()`, disable cancellation propagation to prevent a cyclic
cancellation.
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 573d87f..210c876 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,5 @@
+## 2.12.0-dev
+
## 2.11.0
* Add `CancelableOperation.fromValue`.
diff --git a/lib/src/cancelable_operation.dart b/lib/src/cancelable_operation.dart
index 09997e0..8cc0c67 100644
--- a/lib/src/cancelable_operation.dart
+++ b/lib/src/cancelable_operation.dart
@@ -4,6 +4,8 @@
import 'dart:async';
+import 'package:collection/collection.dart';
+
/// An asynchronous operation that can be cancelled.
///
/// The value of this operation is exposed as [value]. When this operation is
@@ -81,7 +83,10 @@
// they're not actually cancelled by this.
Future<void> cancelAll() {
done = true;
- return Future.wait(operations.map((operation) => operation.cancel()));
+ return Future.wait([
+ for (var operation in operations)
+ if (!operation.isCanceled) operation.cancel()
+ ]);
}
var completer = CancelableCompleter<T>(onCancel: cancelAll);
@@ -93,7 +98,7 @@
cancelAll()
.whenComplete(() => completer.completeError(error, stackTrace));
}
- });
+ }, propagateCancel: false);
}
return completer.operation;
@@ -224,8 +229,8 @@
onError,
FutureOr<void> Function(CancelableCompleter<R>)? onCancel,
bool propagateCancel = true}) {
- final completer =
- CancelableCompleter<R>(onCancel: propagateCancel ? cancel : null);
+ final completer = CancelableCompleter<R>(
+ onCancel: propagateCancel ? _cancelIfNotCanceled : null);
// if `_completer._inner` completes before `completer` is cancelled
// call `onValue` or `onError` with the result, and complete `completer`
@@ -258,20 +263,16 @@
try {
await onError(error, stack, completer);
} catch (error2, stack2) {
- completer.completeError(
+ completer.completeErrorIfPending(
error2, identical(error, error2) ? stack : stack2);
}
});
- _completer._cancelCompleter?.future.whenComplete(onCancel == null
- ? completer._cancel
- : () async {
- if (completer.isCanceled) return;
- try {
- await onCancel(completer);
- } catch (error, stack) {
- completer.completeError(error, stack);
- }
- });
+ final cancelForwarder = _CancelForwarder<R>(completer, onCancel);
+ if (_completer.isCanceled) {
+ cancelForwarder._forward();
+ } else {
+ (_completer._cancelForwarders ??= []).add(cancelForwarder);
+ }
return completer.operation;
}
@@ -281,6 +282,8 @@
/// Returns the result of the `onCancel` callback, if one exists.
Future cancel() => _completer._cancel();
+ Future<void>? _cancelIfNotCanceled() => isCanceled ? null : cancel();
+
/// Whether this operation has been canceled before it completed.
bool get isCanceled => _completer._isCanceled;
@@ -337,11 +340,18 @@
/// ever complete.
/// Set to `null` when [_inner] is completed, because then it's
/// guaranteed that this completer will never complete.
- Completer<void>? _cancelCompleter = Completer<void>();
+ Completer<Object?>? _cancelCompleter = Completer<Object?>();
/// The callback to call if the operation is canceled.
final FutureOr<void> Function()? _onCancel;
+ /// Additional cancellations to forward during cancel.
+ ///
+ /// When a cancelable operation is chained through `then` or `thenOperation` a
+ /// cancellation on the original operation will synchronously cancel the
+ /// chained operations.
+ List<_CancelForwarder>? _cancelForwarders;
+
/// Whether [complete] or [completeError] may still be called.
///
/// Set to false when calling either.
@@ -497,9 +507,59 @@
if (_inner != null) {
_inner = null;
- var onCancel = _onCancel;
- cancelCompleter.complete(onCancel == null ? null : Future.sync(onCancel));
+ cancelCompleter.complete(_invokeCancelCallbacks());
}
return cancelCompleter.future;
}
+
+ /// Invoke [_onCancel] and forward to other completers in [_cancelForwarders].
+ ///
+ /// Returns the same value as [_onCancel]. Legacy uses may return a value
+ /// despite the signature having `void` return.
+ Future<Object?> _invokeCancelCallbacks() async {
+ final FutureOr<Object?> toReturn = _onCancel?.call();
+ final isFuture = toReturn is Future;
+ final cancelFutures = <Future<Object?>>[
+ if (isFuture) toReturn,
+ ...?_cancelForwarders?.map(_forward).whereNotNull()
+ ];
+ final results = (isFuture && cancelFutures.length == 1)
+ ? [await toReturn]
+ : cancelFutures.isNotEmpty
+ ? await Future.wait(cancelFutures)
+ : const [];
+ return isFuture ? results.first : toReturn;
+ }
+}
+
+class _CancelForwarder<T> {
+ final CancelableCompleter<T> completer;
+ final FutureOr<void> Function(CancelableCompleter<T>)? onCancel;
+ _CancelForwarder(this.completer, this.onCancel);
+
+ Future<void>? _forward() {
+ if (completer.isCanceled) return null;
+ final onCancel = this.onCancel;
+ if (onCancel == null) return completer._cancel();
+ try {
+ final result = onCancel(completer);
+ if (result is Future) {
+ return result.catchError(completer.completeErrorIfPending);
+ }
+ } catch (error, stack) {
+ completer.completeErrorIfPending(error, stack);
+ }
+ return null;
+ }
+}
+
+// Helper function to avoid a closure for `List<_CancelForwarder>.map`.
+Future<void>? _forward(_CancelForwarder<Object?> forwarder) =>
+ forwarder._forward();
+
+extension on CancelableCompleter {
+ void completeErrorIfPending(Object error, StackTrace stackTrace) {
+ if (isCompleted) return;
+ completeError(error, stackTrace);
+ }
}
diff --git a/pubspec.yaml b/pubspec.yaml
index 61dd362..9802f78 100644
--- a/pubspec.yaml
+++ b/pubspec.yaml
@@ -1,5 +1,5 @@
name: async
-version: 2.11.0
+version: 2.12.0-dev
description: Utility functions and classes related to the 'dart:async' library.
repository: https://github.com/dart-lang/async
diff --git a/test/cancelable_operation_test.dart b/test/cancelable_operation_test.dart
index 63ba2c3..f72f7e9 100644
--- a/test/cancelable_operation_test.dart
+++ b/test/cancelable_operation_test.dart
@@ -568,6 +568,17 @@
expect(operation.value, completion('foo'));
expect(operation.isCanceled, false);
});
+
+ test('waits for chained cancellation', () async {
+ var completer = CancelableCompleter<void>();
+ var chainedOperation = completer.operation
+ .then((_) => Future.delayed(Duration(milliseconds: 1)))
+ .then((_) => Future.delayed(Duration(milliseconds: 1)));
+
+ await completer.operation.cancel();
+ expect(completer.operation.isCanceled, true);
+ expect(chainedOperation.isCanceled, true);
+ });
});
group('returned operation canceled', () {
