Skip to content

Commit

Permalink
fix(bufferToggle): handle closingSelector completes immediately
Browse files Browse the repository at this point in the history
relates to ReactiveX#1487
  • Loading branch information
kwonoj committed Mar 27, 2016
1 parent c643efc commit 4a56f82
Show file tree
Hide file tree
Showing 2 changed files with 27 additions and 10 deletions.
12 changes: 12 additions & 0 deletions spec/operators/bufferToggle-spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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);
});
});
25 changes: 15 additions & 10 deletions src/operator/bufferToggle.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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);
}
}
}

Expand Down

0 comments on commit 4a56f82

Please sign in to comment.