[fix,async] exception in async _CancelOnErrorSubscriptionWrapper.onEr…

…ror (#855)
dart-lang · Jan 28, 2025 · 2458e1a · 2458e1a
1 parent 3b35aca
commit 2458e1a
Show file tree

Hide file tree

Showing 6 changed files with 108 additions and 6 deletions.
diff --git a/pkgs/async/CHANGELOG.md b/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.

diff --git a/pkgs/async/lib/src/result/error.dart b/pkgs/async/lib/src/result/error.dart
@@ -47,10 +47,15 @@ class ErrorResult implements Result<Never> {
   /// has to be a function expecting only one argument, which will be called
   /// with only the error.
   void handle(Function errorHandler) {
-    if (errorHandler is ZoneBinaryCallback) {
+    if (errorHandler is dynamic Function(Object, StackTrace)) {
       errorHandler(error, stackTrace);
+    } else if (errorHandler is dynamic Function(Object)) {
+      errorHandler(error);
     } else {
-      (errorHandler as ZoneUnaryCallback)(error);
+      throw ArgumentError(
+        'is neither Function(Object, StackTrace) nor Function(Object)',
+        'errorHandler',
+      );
     }
   }
 

diff --git a/pkgs/async/lib/src/subscription_stream.dart b/pkgs/async/lib/src/subscription_stream.dart
@@ -76,10 +76,10 @@ 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);
         }
       });
     });

diff --git a/pkgs/async/pubspec.yaml b/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

diff --git a/pkgs/async/test/result/result_test.dart b/pkgs/async/test/result/result_test.dart
@@ -281,6 +281,29 @@ void main() {
     expect(called, isTrue);
   });
 
+  test('handle typed binary', () {
+    final result = ErrorResult('error', stack);
+    var called = false;
+    void f(Object error, StackTrace stackTrace) {
+      called = true;
+    }
+
+    result.handle(f);
+    expect(called, isTrue);
+  });
+
+  test('handle typed unary', () {
+    final result = ErrorResult('error', stack);
+    var called = false;
+
+    void f(Object error) {
+      called = true;
+    }
+
+    result.handle(f);
+    expect(called, isTrue);
+  });
+
   test('handle unary and binary', () {
     var result = ErrorResult('error', stack);
     var called = false;

diff --git a/pkgs/async/test/subscription_stream_test.dart b/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* {