fix(bufferToggle): fix disposal of subscriptions when errors occur

Fix bufferToggle operator to appropriately unsubscribe whenever an error happens, either on the
source or on the opening or closing observables.

Author: staltz

src/operator/bufferToggle.ts

@@ -53,6 +53,13 @@ class BufferToggleSubscriber<T, O> extends Subscriber<T> {
   }
 
   _error(err: any) {
+    const contexts = this.contexts;
+    while (contexts.length > 0) {
+      const context = contexts.shift();
+      context.subscription.unsubscribe();
+      context.buffer = null;
+      context.subscription = null;
+    }
     this.contexts = null;
     this.destination.error(err);
   }
@@ -64,7 +71,9 @@ class BufferToggleSubscriber<T, O> extends Subscriber<T> {
       this.destination.next(context.buffer);
       context.subscription.unsubscribe();
       context.buffer = null;
+      context.subscription = null;
     }
+    this.contexts = null;
     this.destination.complete();
   }
 
@@ -74,16 +83,14 @@ class BufferToggleSubscriber<T, O> extends Subscriber<T> {
 
     let closingNotifier = tryCatch(closingSelector)(value);
     if (closingNotifier === errorObject) {
-      const err = closingNotifier.e;
-      this.contexts = null;
-      this.destination.error(err);
+      this._error(closingNotifier.e);
     } else {
       let context = {
         buffer: [],
         subscription: new Subscription()
       };
       contexts.push(context);
-      const subscriber = new BufferClosingNotifierSubscriber(this, context);
+      const subscriber = new BufferToggleClosingsSubscriber(this, context);
       const subscription = closingNotifier._subscribe(subscriber);
       this.add(context.subscription.add(subscription));
     }
@@ -102,39 +109,40 @@ class BufferToggleSubscriber<T, O> extends Subscriber<T> {
   }
 }
 
-class BufferClosingNotifierSubscriber<T> extends Subscriber<T> {
-  constructor(private parent: BufferToggleSubscriber<any, T>, private context: { subscription: any, buffer: T[] }) {
+class BufferToggleOpeningsSubscriber<T> extends Subscriber<T> {
+  constructor(private parent: BufferToggleSubscriber<any, T>) {
     super(null);
   }
 
-  _next() {
-    this.parent.closeBuffer(this.context);
+  _next(value: T) {
+    this.parent.openBuffer(value);
   }
 
   _error(err) {
     this.parent.error(err);
   }
 
   _complete() {
-    this.parent.closeBuffer(this.context);
+    // noop
   }
 }
 
-class BufferToggleOpeningsSubscriber<T> extends Subscriber<T> {
-  constructor(private parent: BufferToggleSubscriber<any, T>) {
+class BufferToggleClosingsSubscriber<T> extends Subscriber<T> {
+  constructor(private parent: BufferToggleSubscriber<any, T>,
+              private context: { subscription: any, buffer: T[] }) {
     super(null);
   }
 
-  _next(value: T) {
-    this.parent.openBuffer(value);
+  _next() {
+    this.parent.closeBuffer(this.context);
   }
 
   _error(err) {
     this.parent.error(err);
   }
 
   _complete() {
-    // noop
+    this.parent.closeBuffer(this.context);
   }
 }