test added. formatted.

Author: PROGrand

pkgs/async/lib/src/subscription_stream.dart

@@ -76,12 +76,20 @@ 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<void, Object, StackTrace> || handleError is ZoneBinaryCallback) {
-          handleError?.call(error, stackTrace);
-        } else if (handleError is ZoneUnaryCallback<void, Object> || handleError is ZoneUnaryCallback) {
-          handleError?.call(error);
+        if (handleError is ZoneBinaryCallback<void, Object, StackTrace>) {
+          handleError(error, stackTrace);
+        } else if (handleError is ZoneBinaryCallback) {
+          handleError(error, stackTrace);
+        } else if (handleError is ZoneUnaryCallback<void, Object>) {
+          handleError(error);
+        } else if (handleError is ZoneUnaryCallback) {
+          handleError(error);
         }
       });
     });
   }
 }
+
+abstract class BinaryWrapper {
+  void binaryCall<T1, T2>(T1 t1, T2 t2);
+}

pkgs/async/test/subscription_stream_test.dart

@@ -148,8 +148,80 @@ 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();
+    });
+  });
 }
 
+typedef BinaryFunc = void Function(Object s, StackTrace tr);
+
 Stream<int> createStream() async* {
   yield 1;
   await flushMicrotasks();