Teardown first

packages/observable/src/observable.ts

@@ -220,12 +220,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;
 }
 
 /**
@@ -248,8 +242,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.
@@ -283,7 +275,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
@@ -355,7 +346,6 @@ export class Subscriber<T> extends Subscription implements Observer<T> {
     if (!this.closed) {
       this.isStopped = true;
       super.unsubscribe();
-      this._onFinalize?.();
     }
   }
 
@@ -364,19 +354,13 @@ export class Subscriber<T> extends Subscription implements Observer<T> {
   }
 
   protected _error(err: any): void {
-    try {
-      this.destination.error(err);
-    } finally {
-      this.unsubscribe();
-    }
+    this.unsubscribe();
+    this.destination.error(err);
   }
 
   protected _complete(): void {
-    try {
-      this.destination.complete();
-    } finally {
-      this.unsubscribe();
-    }
+    this.unsubscribe();
+    this.destination.complete();
   }
 }
 
@@ -428,22 +412,20 @@ function overrideNext<T>(this: Subscriber<T>, value: T): void {
 }
 
 function overrideError(this: Subscriber<unknown>, err: any): void {
+  this.unsubscribe();
   try {
     this._errorOverride!(err);
   } catch (error) {
     this.destination.error(error);
-  } finally {
-    this.unsubscribe();
   }
 }
 
 function overrideComplete(this: Subscriber<unknown>): void {
+  this.unsubscribe();
   try {
     this._completeOverride!();
   } catch (error) {
     this.destination.error(error);
-  } finally {
-    this.unsubscribe();
   }
 }
 

packages/rxjs/spec/observables/dom/webSocket-spec.ts

@@ -669,7 +669,7 @@ describe('webSocket', () => {
       });
       socket.triggerClose({ wasClean: true });
 
-      expect(results).to.deep.equal(['A next', 'A unsub', 'B next', 'B next', 'B next', 'B complete', 'B unsub']);
+      expect(results).to.deep.equal(['A next', 'A unsub', 'B next', 'B next', 'B next', 'B complete']);
     });
 
     it('should not close the socket until all subscriptions complete', () => {

packages/rxjs/src/internal/observable/dom/WebSocketSubject.ts

@@ -1,4 +1,4 @@
-import type { Subscriber, Subscription} from '@rxjs/observable';
+import type { Subscriber, Subscription } from '@rxjs/observable';
 import { Observable, operate } from '@rxjs/observable';
 import type { NextObserver } from '../../types.js';
 
@@ -219,8 +219,11 @@ export class WebSocketSubject<In, Out = In> extends Observable<Out> {
   multiplex(subMsg: () => In, unsubMsg: () => In, messageFilter: (value: Out) => boolean) {
     return new Observable<Out>((destination) => {
       this.next(subMsg());
+      let isUnsub = true;
       destination.add(() => {
-        this.next(unsubMsg());
+        if (isUnsub) {
+          this.next(unsubMsg());
+        }
       });
       this.subscribe(
         operate({
@@ -230,6 +233,14 @@ export class WebSocketSubject<In, Out = In> extends Observable<Out> {
               destination.next(x);
             }
           },
+          error: (err) => {
+            isUnsub = false;
+            destination.error(err);
+          },
+          complete: () => {
+            isUnsub = false;
+            destination.complete();
+          },
         })
       );
     });

packages/rxjs/src/internal/observable/forkJoin.ts

@@ -165,9 +165,9 @@ export function forkJoin(...args: any[]): Observable<any> {
             }
             values[sourceIndex] = value;
           },
-          complete: () => remainingCompletions--,
-          finalize: () => {
-            if (!remainingCompletions || !hasValue) {
+          complete: () => {
+            remainingCompletions--;
+            if (remainingCompletions <= 0 || !hasValue) {
               if (remainingEmissions === 0) {
                 destination.next(keys ? createObject(keys, values) : values);
                 destination.complete();

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(() => {
+        // Clean up our memory when we finalize
+        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!;
-          },
         })
       );
     });

packages/rxjs/src/internal/operators/bufferTime.ts

@@ -83,6 +83,10 @@ 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(() => {
+        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)
@@ -127,36 +131,33 @@ export function bufferTime<T>(bufferTimeSpan: number, ...otherArgs: any[]): Oper
 
       startBuffer();
 
-      const bufferTimeSubscriber = operate({
-        destination,
-        next: (value: T) => {
-          // Copy the records, so if we need to remove one we
-          // don't mutate the array. It's hard, but not impossible to
-          // set up a buffer time that could mutate the array and
-          // cause issues here.
-          const recordsCopy = bufferRecords!.slice();
-          for (const record of recordsCopy) {
-            // Loop over all buffers and
-            const { buffer } = record;
-            buffer.push(value);
-            // If the buffer is over the max size, we need to emit it.
-            maxBufferSize <= buffer.length && emit(record);
-          }
-        },
-        complete: () => {
-          // The source completed, emit all of the active
-          // buffers we have before we complete.
-          while (bufferRecords?.length) {
-            destination.next(bufferRecords.shift()!.buffer);
-          }
-          bufferTimeSubscriber?.unsubscribe();
-          destination.complete();
-          destination.unsubscribe();
-        },
-        // Clean up
-        finalize: () => (bufferRecords = null),
-      });
-
-      source.subscribe(bufferTimeSubscriber);
+      source.subscribe(
+        operate({
+          destination,
+          next: (value: T) => {
+            // Copy the records, so if we need to remove one we
+            // don't mutate the array. It's hard, but not impossible to
+            // set up a buffer time that could mutate the array and
+            // cause issues here.
+            const recordsCopy = bufferRecords!.slice();
+            for (const record of recordsCopy) {
+              // Loop over all buffers and
+              const { buffer } = record;
+              buffer.push(value);
+              // If the buffer is over the max size, we need to emit it.
+              maxBufferSize <= buffer.length && emit(record);
+            }
+          },
+          complete: () => {
+            // The source completed, emit all of the active
+            // buffers we have before we complete.
+            while (bufferRecords?.length) {
+              destination.next(bufferRecords.shift()!.buffer);
+            }
+            destination.complete();
+            destination.unsubscribe();
+          },
+        })
+      );
     });
 }

packages/rxjs/src/internal/operators/bufferWhen.ts

@@ -1,4 +1,4 @@
-import type { Subscriber} from '@rxjs/observable';
+import type { Subscriber } from '@rxjs/observable';
 import { operate, Observable, from } from '@rxjs/observable';
 import type { ObservableInput, OperatorFunction } from '../types.js';
 import { noop } from '../util/noop.js';
@@ -43,14 +43,18 @@ 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(() => {
+        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 +66,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 +84,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!),
         })
       );
     });

packages/rxjs/src/internal/operators/debounce.ts

@@ -1,4 +1,4 @@
-import type { Subscriber} from '@rxjs/observable';
+import type { Subscriber } from '@rxjs/observable';
 import { operate, Observable, from } from '@rxjs/observable';
 import type { MonoTypeOperatorFunction, ObservableInput } from '../types.js';
 import { noop } from '../util/noop.js';
@@ -69,6 +69,10 @@ 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(() => {
+        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 +110,6 @@ export function debounce<T>(durationSelector: (value: T) => ObservableInput<any>
             emit();
             destination.complete();
           },
-          finalize: () => {
-            // Finalization.
-            lastValue = durationSubscriber = null;
-          },
         })
       );
     });

packages/rxjs/src/internal/operators/debounceTime.ts

@@ -66,6 +66,10 @@ export function debounceTime<T>(dueTime: number, scheduler: SchedulerLike = asyn
       let lastValue: T;
       let activeTask: Subscription | void;
 
+      destination.add(() => {
+        lastValue = activeTask = null!;
+      });
+
       source.subscribe(
         operate({
           destination,
@@ -94,10 +98,6 @@ export function debounceTime<T>(dueTime: number, scheduler: SchedulerLike = asyn
             }
             destination.complete();
           },
-          finalize: () => {
-            // Finalization.
-            lastValue = activeTask = null!;
-          },
         })
       );
     });

packages/rxjs/src/internal/operators/groupBy.ts

@@ -153,6 +153,11 @@ export function groupBy<T, K, R>(
       // A lookup for the groups that we have so far.
       const groups = new Map<K, SubjectLike<any>>();
 
+      destination.add(() => {
+        // Free up memory.
+        groups.clear();
+      });
+
       // Used for notifying all groups and the subscriber in the same way.
       const notify = (cb: (group: Observer<any>) => void) => {
         groups.forEach(cb);
@@ -202,9 +207,18 @@ export function groupBy<T, K, R>(
                     // Our duration notified! We can complete the group.
                     // The group will be removed from the map in the finalization phase.
                     group!.complete();
+                    groups.delete(key);
+                    durationSubscriber?.unsubscribe();
+                  },
+                  error: (err) => {
+                    group!.error(err);
+                    groups.delete(key);
+                    durationSubscriber?.unsubscribe();
+                  },
+                  complete: () => {
+                    groups.delete(key);
                     durationSubscriber?.unsubscribe();
                   },
-                  finalize: () => groups.delete(key),
                 });
 
                 // Start our duration notifier.
@@ -222,11 +236,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