Refactor a callback to async/await (dart-lang/stream_transform#139)
Use a try/finally block over using `whenComplete`. This also allows
combining the async and sync error handling with a single `catch`
instead of using `catch` for sync errors and `catchError` for async
errors.
Switch the early return to the shorter case of immediately listening on
the first stream, and leaving the more complex logic (and more common
case) with one less level of nesting.
This does move the call to `outerSubscription.pause()` to come before
the all to `currentSubscription.cancel()`, but this ordering should not
be meaningful even though it can be observed.
Fix a TODO about using `Future.ignore()`.
diff --git a/pkgs/stream_transform/lib/src/switch.dart b/pkgs/stream_transform/lib/src/switch.dart
index 53a8394..1fa07da 100644
--- a/pkgs/stream_transform/lib/src/switch.dart
+++ b/pkgs/stream_transform/lib/src/switch.dart
@@ -81,36 +81,31 @@
outerStreamDone = true;
if (innerSubscription == null) controller.close();
});
- outerSubscription.onData((innerStream) {
+ outerSubscription.onData((innerStream) async {
var currentSubscription = innerSubscription;
- if (currentSubscription != null) {
- innerSubscription = null;
- try {
- currentSubscription.cancel().catchError(addError).whenComplete(() {
- if (!isBroadcast && !controller.hasListener) {
- // Result single-subscription stream subscription was cancelled
- // while waiting for previous innerStream cancel.
- //
- // Ensure that the last received stream is also listened to and
- // cancelled, then do nothing further.
- // TODO(lrn): When SDK 2.14 is available, use `.ignore()`.
- innerStream
- .listen(null)
- .cancel()
- .then(_ignore, onError: _ignore);
- return;
- }
- outerSubscription.resume();
- listenToInnerStream(innerStream);
- });
- outerSubscription.pause();
- return;
- } catch (error, stack) {
- // The cancel call threw synchronously.
- controller.addError(error, stack);
+ if (currentSubscription == null) {
+ listenToInnerStream(innerStream);
+ return;
+ }
+ innerSubscription = null;
+ outerSubscription.pause();
+ try {
+ await currentSubscription.cancel();
+ } catch (error, stack) {
+ controller.addError(error, stack);
+ } finally {
+ if (!isBroadcast && !controller.hasListener) {
+ // Result single-subscription stream subscription was cancelled
+ // while waiting for previous innerStream cancel.
+ //
+ // Ensure that the last received stream is also listened to and
+ // cancelled, then do nothing further.
+ innerStream.listen(null).cancel().ignore();
+ } else {
+ outerSubscription.resume();
+ listenToInnerStream(innerStream);
}
}
- listenToInnerStream(innerStream);
});
if (!isBroadcast) {
controller
