Skip to content

Commit

Permalink
fix(delayWhen): correctly handle synchronous duration observable
Browse files Browse the repository at this point in the history 
- closes ReactiveX#2587
  • Loading branch information
@kwonoj
kwonoj committed May 3, 2017
1 parent 83ebe90 commit d3d0371
Show file tree
Hide file tree
Showing 2 changed files with 86 additions and 71 deletions.
146 changes: 79 additions & 67 deletions spec/operators/delayWhen-spec.ts
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import * as Rx from '../../dist/cjs/Rx';
import marbleTestingSignature = require('../helpers/marble-testing'); // tslint:disable-line:no-require-imports

import { expect } from 'chai';
declare const { asDiagram };
declare const hot: typeof marbleTestingSignature.hot;
declare const cold: typeof marbleTestingSignature.cold;
Expand All @@ -11,15 +11,15 @@ declare const rxTestScheduler: Rx.TestScheduler;
/** @test {delayWhen} */
describe('Observable.prototype.delayWhen', () => {
asDiagram('delayWhen(durationSelector)')('should delay by duration selector', () => {
const e1 = hot('---a---b---c--|');
const expected = '-----a------c----(b|)';
const subs = '^ !';
const selector = [cold( '--x--|'),
cold( '----------(x|)'),
cold( '-x--|')];
const e1 = hot('---a---b---c--|');
const expected = '-----a------c----(b|)';
const subs = '^ !';
const selector = [cold('--x--|'),
cold('----------(x|)'),
cold('-x--|')];
const selectorSubs = [' ^ ! ',
' ^ !',
' ^! '];
' ^ !',
' ^! '];

let idx = 0;
function durationSelector(x) {
Expand All @@ -36,12 +36,12 @@ describe('Observable.prototype.delayWhen', () => {
});

it('should delay by selector', () => {
const e1 = hot('--a--b--|');
const expected = '---a--b-|';
const subs = '^ !';
const selector = cold( '-x--|');
const e1 = hot('--a--b--|');
const expected = '---a--b-|';
const subs = '^ !';
const selector = cold('-x--|');
const selectorSubs = [' ^! ',
' ^! '];
' ^! '];

const result = e1.delayWhen((x: any) => selector);

Expand All @@ -51,11 +51,11 @@ describe('Observable.prototype.delayWhen', () => {
});

it('should raise error if source raises error', () => {
const e1 = hot('--a--#');
const expected = '---a-#';
const subs = '^ !';
const selector = cold( '-x--|');
const selectorSubs = ' ^! ';
const e1 = hot('--a--#');
const expected = '---a-#';
const subs = '^ !';
const selector = cold('-x--|');
const selectorSubs = ' ^! ';

const result = e1.delayWhen((x: any) => selector);

Expand All @@ -65,11 +65,11 @@ describe('Observable.prototype.delayWhen', () => {
});

it('should raise error if selector raises error', () => {
const e1 = hot('--a--b--|');
const expected = '---#';
const subs = '^ !';
const selector = cold( '-#');
const selectorSubs = ' ^! ';
const e1 = hot('--a--b--|');
const expected = '---#';
const subs = '^ !';
const selector = cold('-#');
const selectorSubs = ' ^! ';

const result = e1.delayWhen((x: any) => selector);

Expand All @@ -79,12 +79,12 @@ describe('Observable.prototype.delayWhen', () => {
});

it('should delay by selector and completes after value emits', () => {
const e1 = hot('--a--b--|');
const expected = '---------a--(b|)';
const subs = '^ !';
const e1 = hot('--a--b--|');
const expected = '---------a--(b|)';
const subs = '^ !';
const selector = cold('-------x--|');
const selectorSubs = [' ^ !',
' ^ !'];
' ^ !'];

const result = e1.delayWhen((x: any) => selector);

Expand All @@ -94,12 +94,12 @@ describe('Observable.prototype.delayWhen', () => {
});

it('should delay by selector completes if selector does not emits', () => {
const e1 = hot('--a--b--|');
const expected = '------a--(b|)';
const subs = '^ !';
const selector = cold( '----|');
const e1 = hot('--a--b--|');
const expected = '------a--(b|)';
const subs = '^ !';
const selector = cold('----|');
const selectorSubs = [' ^ !',
' ^ !'];
' ^ !'];

const result = e1.delayWhen((x: any) => selector);

Expand All @@ -109,12 +109,12 @@ describe('Observable.prototype.delayWhen', () => {
});

it('should not emit if selector never emits', () => {
const e1 = hot('--a--b--|');
const expected = '-';
const subs = '^ ';
const selector = cold( '-');
const e1 = hot('--a--b--|');
const expected = '-';
const subs = '^ ';
const selector = cold('-');
const selectorSubs = [' ^ ',
' ^ '];
' ^ '];

const result = e1.delayWhen((x: any) => selector);

Expand All @@ -124,12 +124,12 @@ describe('Observable.prototype.delayWhen', () => {
});

it('should delay by first value from selector', () => {
const e1 = hot('--a--b--|');
const expected = '------a--(b|)';
const subs = '^ !';
const selector = cold( '----x--y--|');
const e1 = hot('--a--b--|');
const expected = '------a--(b|)';
const subs = '^ !';
const selector = cold('----x--y--|');
const selectorSubs = [' ^ !',
' ^ !'];
' ^ !'];

const result = e1.delayWhen((x: any) => selector);

Expand All @@ -139,12 +139,12 @@ describe('Observable.prototype.delayWhen', () => {
});

it('should delay by selector does not completes', () => {
const e1 = hot('--a--b--|');
const expected = '------a--(b|)';
const subs = '^ !';
const selector = cold( '----x-----y---');
const e1 = hot('--a--b--|');
const expected = '------a--(b|)';
const subs = '^ !';
const selector = cold('----x-----y---');
const selectorSubs = [' ^ !',
' ^ !'];
' ^ !'];

const result = e1.delayWhen((x: any) => selector);

Expand All @@ -154,9 +154,9 @@ describe('Observable.prototype.delayWhen', () => {
});

it('should raise error if selector throws', () => {
const e1 = hot('--a--b--|');
const expected = '--#';
const subs = '^ !';
const e1 = hot('--a--b--|');
const expected = '--#';
const subs = '^ !';

const err = new Error('error');
const result = e1.delayWhen(<any>((x: any) => { throw err; }));
Expand All @@ -166,14 +166,14 @@ describe('Observable.prototype.delayWhen', () => {
});

it('should start subscription when subscription delay emits', () => {
const e1 = hot('-----a---b---|');
const expected = ' -----a---b-|';
const subs = ' ^ !';
const selector = cold( '--x--|');
const e1 = hot('-----a---b---|');
const expected = ' -----a---b-|';
const subs = ' ^ !';
const selector = cold('--x--|');
const selectorSubs = [' ^ !',
' ^ !'];
' ^ !'];
const subDelay = cold('--x--|');
const subDelaySub = '^ !';
const subDelaySub = '^ !';

const result = e1.delayWhen((x: any) => selector, subDelay);

Expand All @@ -184,14 +184,14 @@ describe('Observable.prototype.delayWhen', () => {
});

it('should start subscription when subscription delay completes without emit value', () => {
const e1 = hot('-----a---b---|');
const expected = ' -----a---b-|';
const subs = ' ^ !';
const selector = cold( '--x--|');
const e1 = hot('-----a---b---|');
const expected = ' -----a---b-|';
const subs = ' ^ !';
const selector = cold('--x--|');
const selectorSubs = [' ^ !',
' ^ !'];
' ^ !'];
const subDelay = cold('--|');
const subDelaySub = '^ !';
const subDelaySub = '^ !';

const result = e1.delayWhen((x: any) => selector, subDelay);

Expand All @@ -202,11 +202,11 @@ describe('Observable.prototype.delayWhen', () => {
});

it('should raise error when subscription delay raises error', () => {
const e1 = hot('-----a---b---|');
const expected = ' # ';
const selector = cold( '--x--|');
const e1 = hot('-----a---b---|');
const expected = ' # ';
const selector = cold('--x--|');
const subDelay = cold('---#');
const subDelaySub = '^ !';
const subDelaySub = '^ !';

const result = e1.delayWhen((x: any) => selector, subDelay);

Expand All @@ -215,4 +215,16 @@ describe('Observable.prototype.delayWhen', () => {
expectSubscriptions(selector.subscriptions).toBe([]);
expectSubscriptions(subDelay.subscriptions).toBe(subDelaySub);
});

it('should complete when duration selector returns synchronous observable', () => {
let next: boolean = false;
let complete: boolean = false;

Rx.Observable.of(1)
.delayWhen(() => Rx.Observable.of(2))
.subscribe(() => next = true, null, () => complete = true);

expect(next).to.be.true;
expect(complete).to.be.true;
});
});
11 changes: 7 additions & 4 deletions src/operator/delayWhen.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ export function delayWhen<T>(this: Observable<T>, delayDurationSelector: (value:
subscriptionDelay?: Observable<any>): Observable<T> {
if (subscriptionDelay) {
return new SubscriptionDelayObservable(this, subscriptionDelay)
.lift(new DelayWhenOperator(delayDurationSelector));
.lift(new DelayWhenOperator(delayDurationSelector));
}
return this.lift(new DelayWhenOperator(delayDurationSelector));
}
Expand Down Expand Up @@ -112,7 +112,7 @@ class DelayWhenSubscriber<T, R> extends OuterSubscriber<T, R> {
this.tryDelay(delayNotifier, value);
}
} catch (err) {
this.destination.error(err);
this.destination.error(err);
}
}

Expand All @@ -138,9 +138,12 @@ class DelayWhenSubscriber<T, R> extends OuterSubscriber<T, R> {

private tryDelay(delayNotifier: Observable<any>, value: T): void {
const notifierSubscription = subscribeToResult(this, delayNotifier, value);
this.add(notifierSubscription);

this.delayNotifierSubscriptions.push(notifierSubscription);
if (notifierSubscription && !notifierSubscription.closed) {
this.add(notifierSubscription);
this.delayNotifierSubscriptions.push(notifierSubscription);
}

this.values.push(value);
}

Expand Down

0 comments on commit d3d0371

Please sign in to comment.