refactor(OperatorSubscriber): Move error handler arg (#6198)

* refactor(OperatorSubscriber): Move error handler arg

Moves the error handler arg, because in most cases we were passing `undefined` and then a complete handler

* chore: remove pointless comment

src/internal/observable/ConnectableObservable.ts

@@ -66,14 +66,14 @@ export class ConnectableObservable<T> extends Observable<T> {
           new OperatorSubscriber(
             subject as any,
             undefined,
-            (err) => {
-              this._teardown();
-              subject.error(err);
-            },
             () => {
               this._teardown();
               subject.complete();
             },
+            (err) => {
+              this._teardown();
+              subject.error(err);
+            },
             () => this._teardown()
           )
         )

src/internal/observable/combineLatest.ts

@@ -251,7 +251,6 @@ export function combineLatestInit(
                       subscriber.next(valueTransform(values.slice()));
                     }
                   },
-                  undefined,
                   () => {
                     if (!--active) {
                       // We only complete the result if we have no more active

src/internal/observable/dom/fetch.ts

@@ -154,13 +154,12 @@ export function fromFetch<T>(
               subscriber,
               // Values are passed through to the subscriber
               undefined,
-              // Error handling
-              handleError,
               // The projected response is complete.
               () => {
                 abortable = false;
                 subscriber.complete();
-              }
+              },
+              handleError
             )
           );
         } else {

src/internal/observable/forkJoin.ts

@@ -152,7 +152,6 @@ export function forkJoin(...args: any[]): Observable<any> {
             }
             values[sourceIndex] = value;
           },
-          undefined,
           () => {
             if (!--remainingCompletions || !hasValue) {
               if (!remainingEmissions) {

src/internal/observable/zip.ts

@@ -95,8 +95,6 @@ export function zip(...args: unknown[]): Observable<unknown> {
                   }
                 }
               },
-              // Any error is passed through the result.
-              undefined,
               () => {
                 // This source completed. Mark it as complete so we can check it later
                 // if we have to.

src/internal/operators/OperatorSubscriber.ts

@@ -20,8 +20,8 @@ export class OperatorSubscriber<T> extends Subscriber<T> {
   constructor(
     destination: Subscriber<any>,
     onNext?: (value: T) => void,
-    onError?: (err: any) => void,
     onComplete?: () => void,
+    onError?: (err: any) => void,
     private onFinalize?: () => void
   ) {
     // It's important - for performance reasons - that all of this class's

src/internal/operators/audit.ts

@@ -81,11 +81,10 @@ export function audit<T>(durationSelector: (value: T) => ObservableInput<any>):
           lastValue = value;
           if (!durationSubscriber) {
             innerFrom(durationSelector(value)).subscribe(
-              (durationSubscriber = new OperatorSubscriber(subscriber, endDuration, undefined, cleanupDuration))
+              (durationSubscriber = new OperatorSubscriber(subscriber, endDuration, cleanupDuration))
             );
           }
         },
-        undefined,
         () => {
           isComplete = true;
           (!hasValue || !durationSubscriber || durationSubscriber.closed) && subscriber.complete();

src/internal/operators/buffer.ts

@@ -52,8 +52,6 @@ export function buffer<T>(closingNotifier: Observable<any>): OperatorFunction<T,
       new OperatorSubscriber(
         subscriber,
         (value) => currentBuffer.push(value),
-        // Pass all errors to the consumer.
-        undefined,
         () => {
           subscriber.next(currentBuffer);
           subscriber.complete();
@@ -71,8 +69,6 @@ export function buffer<T>(closingNotifier: Observable<any>): OperatorFunction<T,
           currentBuffer = [];
           subscriber.next(b);
         },
-        // Pass all errors to the consumer.
-        undefined,
         noop
       )
     );

src/internal/operators/bufferCount.ts

@@ -102,7 +102,6 @@ export function bufferCount<T>(bufferSize: number, startBufferEvery: number | nu
             }
           }
         },
-        undefined,
         () => {
           // When the source completes, emit all of our
           // active buffers.
@@ -111,6 +110,8 @@ export function bufferCount<T>(bufferSize: number, startBufferEvery: number | nu
           }
           subscriber.complete();
         },
+        // Pass all errors through to consumer.
+        undefined,
         () => {
           // Clean up our memory when we teardown
           buffers = null!;

src/internal/operators/bufferTime.ts

@@ -151,7 +151,6 @@ export function bufferTime<T>(bufferTimeSpan: number, ...otherArgs: any[]): Oper
           maxBufferSize <= buffer.length && emit(record);
         }
       },
-      undefined,
       () => {
         // The source completed, emit all of the active
         // buffers we have before we complete.
@@ -162,6 +161,8 @@ export function bufferTime<T>(bufferTimeSpan: number, ...otherArgs: any[]): Oper
         subscriber.complete();
         subscriber.unsubscribe();
       },
+      // Pass all errors through to consumer.
+      undefined,
       // Clean up
       () => (bufferRecords = null)
     );

src/internal/operators/bufferToggle.ts

@@ -75,11 +75,8 @@ export function bufferToggle<T, O>(
           };
 
           // The line below will add the subscription to the parent subscriber *and* the closing subscription.
-          closingSubscription.add(
-            innerFrom(closingSelector(openValue)).subscribe(new OperatorSubscriber(subscriber, emitBuffer, undefined, noop))
-          );
+          closingSubscription.add(innerFrom(closingSelector(openValue)).subscribe(new OperatorSubscriber(subscriber, emitBuffer, noop)));
         },
-        undefined,
         noop
       )
     );
@@ -93,7 +90,6 @@ export function bufferToggle<T, O>(
             buffer.push(value);
           }
         },
-        undefined,
         () => {
           // Source complete. Emit all pending buffers.
           while (buffers.length > 0) {

src/internal/operators/bufferWhen.ts

@@ -68,7 +68,7 @@ export function bufferWhen<T>(closingSelector: () => ObservableInput<any>): Oper
       b && subscriber.next(b);
 
       // Get a new closing notifier and subscribe to it.
-      innerFrom(closingSelector()).subscribe((closingSubscriber = new OperatorSubscriber(subscriber, openBuffer, undefined, noop)));
+      innerFrom(closingSelector()).subscribe((closingSubscriber = new OperatorSubscriber(subscriber, openBuffer, noop)));
     };
 
     // Start the first buffer.
@@ -80,14 +80,14 @@ export function bufferWhen<T>(closingSelector: () => ObservableInput<any>): Oper
         subscriber,
         // Add every new value to the current buffer.
         (value) => buffer?.push(value),
-        // All errors are passed through to the consumer.
-        undefined,
         // When we complete, emit the buffer if we have one,
         // then complete the result.
         () => {
           buffer && subscriber.next(buffer);
           subscriber.complete();
         },
+        // Pass all errors through to consumer.
+        undefined,
         // Release memory on teardown
         () => (buffer = closingSubscriber = null!)
       )

src/internal/operators/catchError.ts

@@ -112,7 +112,7 @@ export function catchError<T, O extends ObservableInput<any>>(
     let handledResult: Observable<ObservedValueOf<O>>;
 
     innerSub = source.subscribe(
-      new OperatorSubscriber(subscriber, undefined, (err) => {
+      new OperatorSubscriber(subscriber, undefined, undefined, (err) => {
         handledResult = innerFrom(selector(err, catchError(selector)(source)));
         if (innerSub) {
           innerSub.unsubscribe();

src/internal/operators/debounce.ts

@@ -96,17 +96,18 @@ export function debounce<T>(durationSelector: (value: T) => ObservableInput<any>
           lastValue = value;
           // Capture our duration subscriber, so we can unsubscribe it when we're notified
           // and we're going to emit the value.
-          durationSubscriber = new OperatorSubscriber(subscriber, emit, undefined, noop);
+          durationSubscriber = new OperatorSubscriber(subscriber, emit, noop);
           // Subscribe to the duration.
           innerFrom(durationSelector(value)).subscribe(durationSubscriber);
         },
-        undefined,
         () => {
           // Source completed.
           // Emit any pending debounced values then complete
           emit();
           subscriber.complete();
         },
+        // Pass all errors through to consumer
+        undefined,
         () => {
           // Teardown.
           lastValue = durationSubscriber = null;

src/internal/operators/debounceTime.ts

@@ -104,13 +104,14 @@ export function debounceTime<T>(dueTime: number, scheduler: SchedulerLike = asyn
             activeTask = scheduler.schedule(emitWhenIdle, dueTime);
           }
         },
-        undefined,
         () => {
           // Source completed.
           // Emit any pending debounced values then complete
           emit();
           subscriber.complete();
         },
+        // Pass all errors through to consumer.
+        undefined,
         () => {
           // Teardown.
           lastValue = activeTask = null;

src/internal/operators/defaultIfEmpty.ts

@@ -46,7 +46,6 @@ export function defaultIfEmpty<T, R>(defaultValue: R): OperatorFunction<T, T | R
           hasValue = true;
           subscriber.next(value);
         },
-        undefined,
         () => {
           if (!hasValue) {
             subscriber.next(defaultValue!);

src/internal/operators/distinct.ts

@@ -84,6 +84,6 @@ export function distinct<T, K>(keySelector?: (value: T) => K, flushes?: Observab
       })
     );
 
-    flushes?.subscribe(new OperatorSubscriber(subscriber, () => distinctKeys.clear(), undefined, noop));
+    flushes?.subscribe(new OperatorSubscriber(subscriber, () => distinctKeys.clear(), noop));
   });
 }

src/internal/operators/every.ts

@@ -53,7 +53,6 @@ export function every<T>(
             subscriber.complete();
           }
         },
-        undefined,
         () => {
           subscriber.next(true);
           subscriber.complete();

src/internal/operators/exhaustAll.ts

@@ -57,14 +57,13 @@ export function exhaustAll<O extends ObservableInput<any>>(): OperatorFunction<O
         (inner) => {
           if (!innerSub) {
             innerSub = innerFrom(inner).subscribe(
-              new OperatorSubscriber(subscriber, undefined, undefined, () => {
+              new OperatorSubscriber(subscriber, undefined, () => {
                 innerSub = null;
                 isComplete && subscriber.complete();
               })
             );
           }
         },
-        undefined,
         () => {
           isComplete = true;
           !innerSub && subscriber.complete();

src/internal/operators/exhaustMap.ts

@@ -83,14 +83,13 @@ export function exhaustMap<T, R, O extends ObservableInput<any>>(
         subscriber,
         (outerValue) => {
           if (!innerSub) {
-            innerSub = new OperatorSubscriber(subscriber, undefined, undefined, () => {
+            innerSub = new OperatorSubscriber(subscriber, undefined, () => {
               innerSub = null;
               isComplete && subscriber.complete();
             });
             innerFrom(project(outerValue, index++)).subscribe(innerSub);
           }
         },
-        undefined,
         () => {
           isComplete = true;
           !innerSub && subscriber.complete();

src/internal/operators/find.ts

@@ -79,7 +79,6 @@ export function createFind<T>(
             subscriber.complete();
           }
         },
-        undefined,
         () => {
           subscriber.next(findIndex ? -1 : undefined);
           subscriber.complete();

src/internal/operators/groupBy.ts

@@ -175,11 +175,11 @@ export function groupBy<T, K, R>(
                   group!.complete();
                   durationSubscriber?.unsubscribe();
                 },
+                // Completions are also sent to the group, but just the group.
+                undefined,
                 // Errors on the duration subscriber are sent to the group
                 // but only the group. They are not sent to the main subscription.
                 undefined,
-                // Completions are also sent to the group, but just the group.
-                undefined,
                 // Teardown: Remove this group from our map.
                 () => groups.delete(key)
               );
@@ -195,10 +195,10 @@ export function groupBy<T, K, R>(
           handleError(err);
         }
       },
-      // Error from the source.
-      handleError,
       // Source completes.
       () => notify((consumer) => consumer.complete()),
+      // Error from the source.
+      handleError,
       // Free up memory.
       // When the source subscription is _finally_ torn down, release the subjects and keys
       // in our groups Map, they may be quite large and we don't want to keep them around if we

src/internal/operators/isEmpty.ts

@@ -74,7 +74,6 @@ export function isEmpty<T>(): OperatorFunction<T, boolean> {
           subscriber.next(false);
           subscriber.complete();
         },
-        undefined,
         () => {
           subscriber.next(true);
           subscriber.complete();

src/internal/operators/materialize.ts

@@ -65,13 +65,13 @@ export function materialize<T>(): OperatorFunction<T, Notification<T> & Observab
         (value) => {
           subscriber.next(Notification.createNext(value));
         },
-        (err) => {
-          subscriber.next(Notification.createError(err));
-          subscriber.complete();
-        },
         () => {
           subscriber.next(Notification.createComplete());
           subscriber.complete();
+        },
+        (err) => {
+          subscriber.next(Notification.createError(err));
+          subscriber.complete();
         }
       )
     );

src/internal/operators/mergeInternals.ts

@@ -84,13 +84,13 @@ export function mergeInternals<T, R>(
             subscriber.next(innerValue);
           }
         },
-        // Errors are passed to the destination.
-        undefined,
         () => {
           // Flag that we have completed, so we know to check the buffer
           // during finalization.
           innerComplete = true;
         },
+        // Errors are passed to the destination.
+        undefined,
         () => {
           // During finalization, if the inner completed (it wasn't errored or
           // cancelled), then we want to try the next item in the buffer if
@@ -129,17 +129,11 @@ export function mergeInternals<T, R>(
 
   // Subscribe to our source observable.
   source.subscribe(
-    new OperatorSubscriber(
-      subscriber,
-      outerNext,
-      // Errors are passed through
-      undefined,
-      () => {
-        // Outer completed, make a note of it, and check to see if we can complete everything.
-        isComplete = true;
-        checkComplete();
-      }
-    )
+    new OperatorSubscriber(subscriber, outerNext, () => {
+      // Outer completed, make a note of it, and check to see if we can complete everything.
+      isComplete = true;
+      checkComplete();
+    })
   );
 
   // Additional teardown (for when the destination is torn down).

src/internal/operators/observeOn.ts

@@ -60,8 +60,8 @@ export function observeOn<T>(scheduler: SchedulerLike, delay: number = 0): MonoT
       new OperatorSubscriber(
         subscriber,
         (value) => subscriber.add(scheduler.schedule(() => subscriber.next(value), delay)),
-        (err) => subscriber.add(scheduler.schedule(() => subscriber.error(err), delay)),
-        () => subscriber.add(scheduler.schedule(() => subscriber.complete(), delay))
+        () => subscriber.add(scheduler.schedule(() => subscriber.complete(), delay)),
+        (err) => subscriber.add(scheduler.schedule(() => subscriber.error(err), delay))
       )
     );
   });