feat(single): add RxSingles.forkJoin3..forkJoin9 and `RxSingles.for…

…kJoinList` (#80) * feat(single): add `RxSingles.forkJoin3` -> `RxSingles.forkJoin9` and `RxSingles.forkJoinList` * feat(single): add `RxSingles.forkJoin3` -> `RxSingles.forkJoin9` and `RxSingles.forkJoinList` * feat(single): add `RxSingles.forkJoin3` -> `RxSingles.forkJoin9` and `RxSingles.forkJoinList` * feat(single): add `RxSingles.forkJoin3` -> `RxSingles.forkJoin9` and `RxSingles.forkJoinList` * feat(single): add `RxSingles.forkJoin3` -> `RxSingles.forkJoin9` and `RxSingles.forkJoinList` * feat(single): add `RxSingles.forkJoin3` -> `RxSingles.forkJoin9` and `RxSingles.forkJoinList` * feat(single): add `RxSingles.forkJoin3` -> `RxSingles.forkJoin9` and `RxSingles.forkJoinList` * feat(single): add `RxSingles.forkJoin3` -> `RxSingles.forkJoin9` and `RxSingles.forkJoinList` * feat(single): add `RxSingles.forkJoin3` -> `RxSingles.forkJoin9` and `RxSingles.forkJoinList`
hoc081098 · Nov 20, 2022 · 1fddf27 · 1fddf27
1 parent 83610f4
commit 1fddf27
Show file tree

Hide file tree

Showing 3 changed files with 725 additions and 46 deletions.
diff --git a/lib/src/single/internal.dart b/lib/src/single/internal.dart
@@ -18,14 +18,15 @@ extension TakeFirstDataOrFirstErrorExtension<T> on Stream<T> {
     controller.onListen = () {
       subscription = listen(
         (v) {
-          subscription!.cancel();
-          subscription = null;
-
           controller.add(v);
+
+          // Closing also unsubscribes all subscribers, which unsubscribes
+          // this from source.
           controller.close();
         },
         onError: (Object e, StackTrace s) {
-          subscription!.cancel();
+          // cancelOnError=true cause the subscription to be canceled before
+          // the error event is delivered to the listener.
           subscription = null;
 
           controller.addError(e, s);
@@ -35,6 +36,7 @@ extension TakeFirstDataOrFirstErrorExtension<T> on Stream<T> {
           throw APIContractViolationError(
               'Internal API error! Please file a bug at: https://github.com/hoc081098/rxdart_ext/issues/new');
         },
+        cancelOnError: true,
       );
 
       if (!isBroadcast) {