fix(bufferTime): handle closing context when synchronously unsubscribed

closes #1763

Author: kwonoj

spec/operators/bufferTime-spec.ts

@@ -284,4 +284,20 @@ describe('Observable.prototype.bufferTime', () => {
     expectObservable(result).toBe(expected, values);
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
+
+  it('should not throw when subscription synchronously unsubscribed after emit', () => {
+    const e1 =   hot('---a---b---c---d---e---f---g-----|');
+    const subs =     '^                   !';
+    const t = time(  '----------|');
+    const expected = '----------w---------(x|)';
+    const values = {
+      w: ['a', 'b'],
+      x: ['c', 'd', 'e']
+    };
+
+    const result = e1.bufferTime(t, null, Number.POSITIVE_INFINITY, rxTestScheduler).take(2);
+
+    expectObservable(result).toBe(expected, values);
+    expectSubscriptions(e1.subscriptions).toBe(subs);
+  });
 });

src/operator/bufferTime.ts

@@ -183,15 +183,19 @@ class BufferTimeSubscriber<T> extends Subscriber<T> {
   }
 
   openContext(): Context<T> {
-    let context: Context<T> = new Context<T>();
+    const context: Context<T> = new Context<T>();
     this.contexts.push(context);
     return context;
   }
 
   closeContext(context: Context<T>) {
     this.destination.next(context.buffer);
     const contexts = this.contexts;
-    contexts.splice(contexts.indexOf(context), 1);
+
+    const spliceIndex = contexts ? contexts.indexOf(context) : -1;
+    if (spliceIndex >= 0) {
+      contexts.splice(contexts.indexOf(context), 1);
+    }
   }
 }
 
@@ -203,8 +207,8 @@ function dispatchBufferTimeSpanOnly(state: any) {
     subscriber.closeContext(prevContext);
   }
 
-  state.context = subscriber.openContext();
   if (!subscriber.isUnsubscribed) {
+    state.context = subscriber.openContext();
     state.context.closeAction = (<any>this).schedule(state, state.bufferTimeSpan);
   }
 }