fix(Subscriber): don't leak destination (#6116)

* test: rearrange shareReplay test slightly

* test: add failing Subscriber destination leak test

* fix: null destination on unsubscribe

* fix: capture destination before onNext, etc.

spec/Subscriber-spec.ts

@@ -1,6 +1,6 @@
 import { expect } from 'chai';
 import { SafeSubscriber } from 'rxjs/internal/Subscriber';
-import { Subscriber, Observable, config, of } from 'rxjs';
+import { Subscriber, Observable, config, of, Observer } from 'rxjs';
 import { asInteropSubscriber } from './helpers/interop-helper';
 import { getRegisteredTeardowns } from './helpers/subscription';
 
@@ -242,4 +242,30 @@ describe('Subscriber', () => {
       expect(consumer.valuesProcessed).not.to.equal(['new', 'new']);
     });
   });
+
+  const FinalizationRegistry = (global as any).FinalizationRegistry;
+  if (FinalizationRegistry) {
+
+    it('should not leak the destination', (done) => {
+      let observer: Observer<number> | undefined = {
+        next() { /* noop */ },
+        error() { /* noop */ },
+        complete() { /* noop */ }
+      };
+
+      const registry = new FinalizationRegistry((value: any) => {
+        expect(value).to.equal('observer');
+        done();
+      });
+      registry.register(observer, 'observer');
+
+      const subscription = of(42).subscribe(observer);
+
+      observer = undefined;
+      global.gc();
+    });
+
+  } else {
+    console.warn(`No support for FinalizationRegistry in Node ${process.version}`);
+  }
 });

spec/operators/shareReplay-spec.ts

@@ -323,11 +323,12 @@ describe('shareReplay operator', () => {
   if (FinalizationRegistry) {
 
     it('should not leak the subscriber for sync sources', (done) => {
+      let callback: (() => void) | undefined = () => { /* noop */ };
+
       const registry = new FinalizationRegistry((value: any) => {
         expect(value).to.equal('callback');
         done();
       });
-      let callback: (() => void) | undefined = () => { /* noop */ };
       registry.register(callback, 'callback');
 
       const shared = of(42).pipe(shareReplay(1));

src/internal/Subscriber.ts

@@ -105,6 +105,7 @@ export class Subscriber<T> extends Subscription implements Observer<T> {
     if (!this.closed) {
       this.isStopped = true;
       super.unsubscribe();
+      this.destination = null!;
     }
   }
 

src/internal/operators/OperatorSubscriber.ts

@@ -42,7 +42,7 @@ export class OperatorSubscriber<T> extends Subscriber<T> {
           try {
             onNext(value);
           } catch (err) {
-            this.destination.error(err);
+            destination.error(err);
           }
         }
       : super._next;
@@ -52,7 +52,7 @@ export class OperatorSubscriber<T> extends Subscriber<T> {
             onError(err);
           } catch (err) {
             // Send any errors that occur down stream.
-            this.destination.error(err);
+            destination.error(err);
           } finally {
             // Ensure teardown.
             this.unsubscribe();
@@ -65,7 +65,7 @@ export class OperatorSubscriber<T> extends Subscriber<T> {
             onComplete();
           } catch (err) {
             // Send any errors that occur down stream.
-            this.destination.error(err);
+            destination.error(err);
           } finally {
             // Ensure teardown.
             this.unsubscribe();