WIP: many more operators fixed

benlesh · benlesh · commit 3d2cf10f8b6e · 2024-02-28T17:17:11.000-06:00
diff --git a/packages/rxjs/src/internal/Observable.ts b/packages/rxjs/src/internal/Observable.ts
@@ -221,12 +221,6 @@ export interface SubscriberOverrides<T> {
    * will be handled and passed to the destination's `error` method.
    */
   complete?: () => void;
-  /**
-   * If provided, this function will be called after all teardown has occurred
-   * for this {@link Subscriber}. This is generally used for cleanup purposes
-   * during operator development.
-   */
-  finalize?: () => void;
 }
 
 /**
@@ -249,8 +243,6 @@ export class Subscriber<T> extends Subscription implements Observer<T> {
   protected readonly _errorOverride: ((err: any) => void) | null = null;
   /** @internal */
   protected readonly _completeOverride: (() => void) | null = null;
-  /** @internal */
-  protected readonly _onFinalize: (() => void) | null = null;
 
   /**
    * @deprecated Do not create instances of `Subscriber` directly. Use {@link operate} instead.
@@ -284,7 +276,6 @@ export class Subscriber<T> extends Subscription implements Observer<T> {
     this._nextOverride = overrides?.next ?? null;
     this._errorOverride = overrides?.error ?? null;
     this._completeOverride = overrides?.complete ?? null;
-    this._onFinalize = overrides?.finalize ?? null;
 
     // It's important - for performance reasons - that all of this class's
     // members are initialized and that they are always initialized in the same
@@ -356,7 +347,6 @@ export class Subscriber<T> extends Subscription implements Observer<T> {
     if (!this.closed) {
       this.isStopped = true;
       super.unsubscribe();
-      this._onFinalize?.();
     }
   }
 
@@ -433,12 +423,11 @@ function overrideError(this: Subscriber<unknown>, err: any): void {
 }
 
 function overrideComplete(this: Subscriber<unknown>): void {
+  this.unsubscribe();
   try {
     this._completeOverride!();
   } catch (error) {
     this.destination.error(error);
-  } finally {
-    this.unsubscribe();
   }
 }
 
diff --git a/packages/rxjs/src/internal/operators/bufferCount.ts b/packages/rxjs/src/internal/operators/bufferCount.ts
@@ -63,6 +63,11 @@ export function bufferCount<T>(bufferSize: number, startBufferEvery: number | nu
       let buffers: T[][] = [];
       let count = 0;
 
+      destination.add(() => {
+        // Release memory
+        buffers = null!;
+      });
+
       source.subscribe(
         operate({
           destination,
@@ -108,10 +113,6 @@ export function bufferCount<T>(bufferSize: number, startBufferEvery: number | nu
             }
             destination.complete();
           },
-          finalize: () => {
-            // Clean up our memory when we finalize
-            buffers = null!;
-          },
         })
       );
     });
diff --git a/packages/rxjs/src/internal/operators/bufferTime.ts b/packages/rxjs/src/internal/operators/bufferTime.ts
@@ -83,6 +83,11 @@ export function bufferTime<T>(bufferTimeSpan: number, ...otherArgs: any[]): Oper
       // this is only really used for when *just* the buffer time span is passed.
       let restartOnEmit = false;
 
+      destination.add(() => {
+        // Release memory
+        bufferRecords = null;
+      });
+
       /**
        * Does the work of emitting the buffer from the record, ensuring that the
        * record is removed before the emission so reentrant code (from some custom scheduling, perhaps)
@@ -153,8 +158,6 @@ export function bufferTime<T>(bufferTimeSpan: number, ...otherArgs: any[]): Oper
           destination.complete();
           destination.unsubscribe();
         },
-        // Clean up
-        finalize: () => (bufferRecords = null),
       });
 
       source.subscribe(bufferTimeSubscriber);
diff --git a/packages/rxjs/src/internal/operators/bufferWhen.ts b/packages/rxjs/src/internal/operators/bufferWhen.ts
@@ -1,4 +1,4 @@
-import type { Subscriber} from '../Observable.js';
+import type { Subscriber } from '../Observable.js';
 import { operate, Observable, from } from '../Observable.js';
 import type { ObservableInput, OperatorFunction } from '../types.js';
 import { noop } from '../util/noop.js';
@@ -43,14 +43,19 @@ import { noop } from '../util/noop.js';
  */
 export function bufferWhen<T>(closingSelector: () => ObservableInput<any>): OperatorFunction<T, T[]> {
   return (source) =>
-    new Observable((subscriber) => {
+    new Observable((destination) => {
       // The buffer we keep and emit.
       let buffer: T[] | null = null;
       // A reference to the subscriber used to subscribe to
       // the closing notifier. We need to hold this so we can
       // end the subscription after the first notification.
       let closingSubscriber: Subscriber<T> | null = null;
 
+      destination.add(() => {
+        // Release memory.
+        buffer = closingSubscriber = null!;
+      });
+
       // Ends the previous closing notifier subscription, so it
       // terminates after the first emission, then emits
       // the current buffer  if there is one, starts a new buffer, and starts a
@@ -62,12 +67,12 @@ export function bufferWhen<T>(closingSelector: () => ObservableInput<any>): Oper
         // emit the buffer if we have one, and start a new buffer.
         const b = buffer;
         buffer = [];
-        b && subscriber.next(b);
+        b && destination.next(b);
 
         // Get a new closing notifier and subscribe to it.
         from(closingSelector()).subscribe(
           (closingSubscriber = operate({
-            destination: subscriber,
+            destination,
             next: openBuffer,
             complete: noop,
           }))
@@ -80,17 +85,15 @@ export function bufferWhen<T>(closingSelector: () => ObservableInput<any>): Oper
       // Subscribe to our source.
       source.subscribe(
         operate({
-          destination: subscriber,
+          destination,
           // Add every new value to the current buffer.
           next: (value) => buffer?.push(value),
           // When we complete, emit the buffer if we have one,
           // then complete the result.
           complete: () => {
-            buffer && subscriber.next(buffer);
-            subscriber.complete();
+            buffer && destination.next(buffer);
+            destination.complete();
           },
-          // Release memory on finalization
-          finalize: () => (buffer = closingSubscriber = null!),
         })
       );
     });
diff --git a/packages/rxjs/src/internal/operators/debounce.ts b/packages/rxjs/src/internal/operators/debounce.ts
@@ -1,4 +1,4 @@
-import type { Subscriber} from '../Observable.js';
+import type { Subscriber } from '../Observable.js';
 import { operate, Observable, from } from '../Observable.js';
 import type { MonoTypeOperatorFunction, ObservableInput } from '../types.js';
 import { noop } from '../util/noop.js';
@@ -69,6 +69,11 @@ export function debounce<T>(durationSelector: (value: T) => ObservableInput<any>
       // The subscriber/subscription for the current debounce, if there is one.
       let durationSubscriber: Subscriber<any> | null = null;
 
+      destination.add(() => {
+        // Clean up
+        lastValue = durationSubscriber = null;
+      });
+
       const emit = () => {
         // Unsubscribe any current debounce subscription we have,
         // we only cared about the first notification from it, and we
@@ -106,10 +111,6 @@ export function debounce<T>(durationSelector: (value: T) => ObservableInput<any>
             emit();
             destination.complete();
           },
-          finalize: () => {
-            // Finalization.
-            lastValue = durationSubscriber = null;
-          },
         })
       );
     });
diff --git a/packages/rxjs/src/internal/operators/debounceTime.ts b/packages/rxjs/src/internal/operators/debounceTime.ts
@@ -66,6 +66,11 @@ export function debounceTime<T>(dueTime: number, scheduler: SchedulerLike = asyn
       let lastValue: T;
       let activeTask: Subscription | void;
 
+      destination.add(() => {
+        // Clean up
+        lastValue = activeTask = null!;
+      });
+
       source.subscribe(
         operate({
           destination,
@@ -94,10 +99,6 @@ export function debounceTime<T>(dueTime: number, scheduler: SchedulerLike = asyn
             }
             destination.complete();
           },
-          finalize: () => {
-            // Finalization.
-            lastValue = activeTask = null!;
-          },
         })
       );
     });
diff --git a/packages/rxjs/src/internal/operators/groupBy.ts b/packages/rxjs/src/internal/operators/groupBy.ts
@@ -204,7 +204,14 @@ export function groupBy<T, K, R>(
                     group!.complete();
                     durationSubscriber?.unsubscribe();
                   },
-                  finalize: () => groups.delete(key),
+                  error: (err) => {
+                    group!.error(err);
+                    groups.delete(key);
+                  },
+                  complete: () => {
+                    group!.complete();
+                    groups.delete(key);
+                  },
                 });
 
                 // Start our duration notifier.
@@ -222,11 +229,6 @@ export function groupBy<T, K, R>(
         error: handleError,
         // Source completes.
         complete: () => notify((consumer) => consumer.complete()),
-        // 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
-        // don't have to.
-        finalize: () => groups.clear(),
       });
 
       // Subscribe to the source
diff --git a/packages/rxjs/src/internal/operators/mergeInternals.ts b/packages/rxjs/src/internal/operators/mergeInternals.ts
@@ -57,11 +57,6 @@ export function mergeInternals<T, R>(
     // against our concurrency limit later.
     active++;
 
-    // A flag used to show that the inner observable completed.
-    // This is checked during finalization to see if we should
-    // move to the next item in the buffer, if there is on.
-    let innerComplete = false;
-
     // Start our inner subscription.
     from(project(value, index++)).subscribe(
       operate({
@@ -81,36 +76,26 @@ export function mergeInternals<T, R>(
           }
         },
         complete: () => {
-          // Flag that we have completed, so we know to check the buffer
-          // during finalization.
-          innerComplete = true;
-        },
-        finalize: () => {
-          // During finalization, if the inner completed (it wasn't errored or
-          // cancelled), then we want to try the next item in the buffer if
-          // there is one.
-          if (innerComplete) {
-            // We have to wrap this in a try/catch because it happens during
-            // finalization, possibly asynchronously, and we want to pass
-            // any errors that happen (like in a projection function) to
-            // the outer Subscriber.
-            try {
-              // INNER SOURCE COMPLETE
-              // Decrement the active count to ensure that the next time
-              // we try to call `doInnerSub`, the number is accurate.
-              active--;
-              // If we have more values in the buffer, try to process those
-              // Note that this call will increment `active` ahead of the
-              // next conditional, if there were any more inner subscriptions
-              // to start.
-              while (buffer.length && active < concurrent) {
-                doInnerSub(buffer.shift()!);
-              }
-              // Check to see if we can complete, and complete if so.
-              checkComplete();
-            } catch (err) {
-              destination.error(err);
+          // We have to wrap this in a try/catch because it happens during
+          // finalization, possibly asynchronously, and we want to pass
+          // any errors that happen (like in a projection function) to
+          // the outer Subscriber.
+          try {
+            // INNER SOURCE COMPLETE
+            // Decrement the active count to ensure that the next time
+            // we try to call `doInnerSub`, the number is accurate.
+            active--;
+            // If we have more values in the buffer, try to process those
+            // Note that this call will increment `active` ahead of the
+            // next conditional, if there were any more inner subscriptions
+            // to start.
+            while (buffer.length && active < concurrent) {
+              doInnerSub(buffer.shift()!);
             }
+            // Check to see if we can complete, and complete if so.
+            checkComplete();
+          } catch (err) {
+            destination.error(err);
           }
         },
       })
diff --git a/packages/rxjs/src/internal/operators/takeLast.ts b/packages/rxjs/src/internal/operators/takeLast.ts
@@ -50,6 +50,12 @@ export function takeLast<T>(count: number): MonoTypeOperatorFunction<T> {
           let ring = new Array<T>(count);
           // This counter is how we track where we are at in the ring buffer.
           let counter = 0;
+
+          destination.add(() => {
+            // During finalization release the values in our buffer.
+            ring = null!;
+          });
+
           source.subscribe(
             operate({
               destination,
@@ -73,10 +79,6 @@ export function takeLast<T>(count: number): MonoTypeOperatorFunction<T> {
                 // All done. This will also trigger clean up.
                 destination.complete();
               },
-              finalize: () => {
-                // During finalization release the values in our buffer.
-                ring = null!;
-              },
             })
           );
         });
diff --git a/packages/rxjs/src/internal/operators/tap.ts b/packages/rxjs/src/internal/operators/tap.ts
@@ -162,10 +162,18 @@ export function tap<T>(observerOrNext?: Partial<TapObserver<T>> | ((value: T) =>
         new Observable((destination) => {
           tapObserver.subscribe?.();
           let isUnsub = true;
+
+          destination.add(() => {
+            if (isUnsub) {
+              tapObserver.unsubscribe?.();
+            }
+            tapObserver.finalize?.();
+          });
+
           source.subscribe(
             operate({
               destination,
-              next: (value) => {
+              next: (value: T) => {
                 tapObserver.next?.(value);
                 destination.next(value);
               },
@@ -179,12 +187,6 @@ export function tap<T>(observerOrNext?: Partial<TapObserver<T>> | ((value: T) =>
                 tapObserver.complete?.();
                 destination.complete();
               },
-              finalize: () => {
-                if (isUnsub) {
-                  tapObserver.unsubscribe?.();
-                }
-                tapObserver.finalize?.();
-              },
             })
           );
         })
diff --git a/packages/rxjs/src/internal/operators/timeout.ts b/packages/rxjs/src/internal/operators/timeout.ts
@@ -1,7 +1,7 @@
 import { asyncScheduler } from '../scheduler/async.js';
 import type { MonoTypeOperatorFunction, SchedulerLike, OperatorFunction, ObservableInput, ObservedValueOf } from '../types.js';
 import { isValidDate } from '../util/isDate.js';
-import type { Subscription} from '../Observable.js';
+import type { Subscription } from '../Observable.js';
 import { Observable, from, operate } from '../Observable.js';
 import { executeSchedule } from '../util/executeSchedule.js';
 
@@ -340,6 +340,13 @@ export function timeout<T, O extends ObservableInput<any>, M>(
         );
       };
 
+      destination.add(() => {
+        timerSubscription?.unsubscribe();
+        // Be sure not to hold the last value in memory after unsubscription
+        // it could be quite large.
+        lastValue = null;
+      });
+
       originalSourceSubscription = source.subscribe(
         operate({
           destination,
@@ -352,12 +359,6 @@ export function timeout<T, O extends ObservableInput<any>, M>(
             // null | undefined are both < 0. Thanks, JavaScript.
             each! > 0 && startTimer(each!);
           },
-          finalize: () => {
-            timerSubscription?.unsubscribe();
-            // Be sure not to hold the last value in memory after unsubscription
-            // it could be quite large.
-            lastValue = null;
-          },
         })
       );
 
diff --git a/packages/rxjs/src/internal/operators/windowCount.ts b/packages/rxjs/src/internal/operators/windowCount.ts
diff --git a/packages/rxjs/src/internal/operators/windowToggle.ts b/packages/rxjs/src/internal/operators/windowToggle.ts
diff --git a/packages/rxjs/src/internal/operators/windowWhen.ts b/packages/rxjs/src/internal/operators/windowWhen.ts
