fix(bufferToggle): handle closingSelector completes immediately

relates to ReactiveX#1487
kwonoj · Mar 21, 2016 · fa09888 · fa09888
1 parent ab6d826
commit fa09888
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 10 deletions.
diff --git a/spec/operators/bufferToggle-spec.ts b/spec/operators/bufferToggle-spec.ts
@@ -380,4 +380,16 @@ describe('Observable.prototype.bufferToggle', () => {
  done.fail();
  });
  });
+
+ it('should handle empty closing observable', () => {
+ const e1 = hot('--a--^---b---c---d---e---f---g---h------|');
+ const subs = '^ !';
+ const e2 = cold('--x-----------y--------z---| ');
+ const expected = '--l-----------m--------n-----------|';
+
+ const result = e1.bufferToggle(e2, () => Observable.empty());
+
+ expectObservable(result).toBe(expected, {l: [], m: [], n: []});
+ expectSubscriptions(e1.subscriptions).toBe(subs);
+ });
 });
diff --git a/src/operator/bufferToggle.ts b/src/operator/bufferToggle.ts
@@ -137,14 +137,14 @@ class BufferToggleSubscriber<T, O> extends OuterSubscriber<T, O> {
 
  private closeBuffer(context: BufferContext<T>): void {
  const contexts = this.contexts;
- if (contexts === null) {
- return;
+
+ if (contexts && context) {
+ const { buffer, subscription } = context;
+ this.destination.next(buffer);
+ contexts.splice(contexts.indexOf(context), 1);
+ this.remove(subscription);
+ subscription.unsubscribe();
  }
- const { buffer, subscription } = context;
- this.destination.next(buffer);
- contexts.splice(contexts.indexOf(context), 1);
- this.remove(subscription);
- subscription.unsubscribe();
  }
 
  private trySubscribe(closingNotifier: any): void {
@@ -156,10 +156,15 @@ class BufferToggleSubscriber<T, O> extends OuterSubscriber<T, O> {
  contexts.push(context);
 
  const innerSubscription = subscribeToResult(this, closingNotifier, <any>context);
- (<any> innerSubscription).context = context;
 
- this.add(innerSubscription);
- subscription.add(innerSubscription);
+ if (!innerSubscription.isUnsubscribed) {
+ (<any> innerSubscription).context = context;
+
+ this.add(innerSubscription);
+ subscription.add(innerSubscription);
+ } else {
+ this.closeBuffer(context);
+ }
  }
 }