fix(Subscriber): handle memory leaks

spec/operators/bufferWhen-spec.ts

@@ -1,6 +1,6 @@
 import { expect } from 'chai';
-import { of, EMPTY, Observable } from 'rxjs';
-import { bufferWhen, mergeMap, takeWhile, take } from 'rxjs/operators';
+import { of, throwError } from 'rxjs';
+import { bufferWhen, mergeMap, takeWhile } from 'rxjs/operators';
 import { TestScheduler } from 'rxjs/testing';
 import { observableMatcher } from '../helpers/observableMatcher';
 
@@ -198,7 +198,7 @@ describe('bufferWhen operator', () => {
       const result = e1.pipe(
         bufferWhen(() => {
           if (i === 1) {
-            throw 'error';
+            return throwError(() => 'error');
           }
           return closings[i++];
         })

src/internal/Subscriber.ts

@@ -106,6 +106,7 @@ export class Subscriber<T> extends Subscription implements Observer<T> {
     if (!this.closed) {
       this.isStopped = true;
       super.unsubscribe();
+      this.destination = NOOP_OBSERVER;
     }
   }
 
@@ -210,3 +211,13 @@ export const EMPTY_OBSERVER: Readonly<Observer<any>> & { closed: true } = {
   error: defaultErrorHandler,
   complete: noop,
 };
+
+/**
+ * The observer used as a stub for subscriptions that have already been closed.
+ */
+export const NOOP_OBSERVER: Readonly<Observer<any>> & { closed: true } = {
+  closed: true,
+  next: noop,
+  error: noop,
+  complete: noop,
+};

src/internal/Subscription.ts

@@ -66,6 +66,7 @@ export class Subscription implements SubscriptionLike {
       }
 
       const { initialTeardown } = this;
+      this.initialTeardown = undefined;
       if (isFunction(initialTeardown)) {
         try {
           initialTeardown();