[fix,async] exception in async _CancelOnErrorSubscriptionWrapper.onError

pkgs/async/CHANGELOG.md

@@ -1,3 +1,7 @@
+## 2.13.0
+
+- Fix type check and cast in SubscriptionStream's cancelOnError wrapper
+
 ## 2.12.0
 
 - Require Dart 3.4.

pkgs/async/lib/src/subscription_stream.dart

@@ -76,12 +76,16 @@ class _CancelOnErrorSubscriptionWrapper<T>
     super.onError((Object error, StackTrace stackTrace) {
       // Wait for the cancel to complete before sending the error event.
       super.cancel().whenComplete(() {
-        if (handleError is ZoneBinaryCallback) {
+        if (handleError is dynamic Function(Object, StackTrace)) {
           handleError(error, stackTrace);
-        } else if (handleError != null) {
-          (handleError as ZoneUnaryCallback)(error);
+        } else if (handleError is dynamic Function(Object)) {
+          handleError(error);
         }
       });
     });
   }
 }
+
+abstract class BinaryWrapper {
+  void binaryCall<T1, T2>(T1 t1, T2 t2);
+}

pkgs/async/pubspec.yaml

@@ -1,5 +1,5 @@
 name: async
-version: 2.12.0
+version: 2.13.0
 description: Utility functions and classes related to the 'dart:async' library.
 repository: https://github.com/dart-lang/core/tree/main/pkgs/async
 issue_tracker: https://github.com/dart-lang/core/issues?q=is%3Aissue+is%3Aopen+label%3Apackage%3Aasync

pkgs/async/test/subscription_stream_test.dart

@@ -148,6 +148,76 @@ void main() {
       });
     }
   });
+
+  group('subscriptionStream error callback', () {
+    test('binary typed', () async {
+      var completer = Completer<void>();
+      var stream = createErrorStream();
+      var sourceSubscription = stream.listen(null, cancelOnError: true);
+      var subscriptionStream = SubscriptionStream(sourceSubscription);
+
+      void f(Object error, StackTrace stackTrace) {
+        completer.complete();
+      }
+
+      subscriptionStream.listen((_) {},
+          onError: f,
+          onDone: () => throw 'should not happen',
+          cancelOnError: true);
+      await completer.future;
+      await flushMicrotasks();
+    });
+
+    test('binary dynamic', () async {
+      var completer = Completer<void>();
+      var stream = createErrorStream();
+      var sourceSubscription = stream.listen(null, cancelOnError: true);
+      var subscriptionStream = SubscriptionStream(sourceSubscription);
+
+      subscriptionStream.listen((_) {},
+          onError: (error, stackTrace) {
+            completer.complete();
+          },
+          onDone: () => throw 'should not happen',
+          cancelOnError: true);
+      await completer.future;
+      await flushMicrotasks();
+    });
+
+    test('unary typed', () async {
+      var completer = Completer<void>();
+      var stream = createErrorStream();
+      var sourceSubscription = stream.listen(null, cancelOnError: true);
+      var subscriptionStream = SubscriptionStream(sourceSubscription);
+
+      void f(Object error) {
+        completer.complete();
+      }
+
+      subscriptionStream.listen((_) {},
+          onError: f,
+          onDone: () => throw 'should not happen',
+          cancelOnError: true);
+      await completer.future;
+      await flushMicrotasks();
+    });
+
+    test('unary dynamic', () async {
+      var completer = Completer<void>();
+      var stream = createErrorStream();
+      var sourceSubscription = stream.listen(null, cancelOnError: true);
+      var subscriptionStream = SubscriptionStream(sourceSubscription);
+
+      subscriptionStream.listen((_) {},
+          onError: (error) {
+            completer.complete();
+          },
+          onDone: () => throw 'should not happen',
+          cancelOnError: true);
+      await completer.future;
+      await flushMicrotasks();
+    });
+  });
 }
 
 Stream<int> createStream() async* {