fix(delayWhen): Emit source value if duration selector completes synchronously

spec/operators/delayWhen-spec.ts

@@ -1,4 +1,4 @@
-import { of } from 'rxjs';
+import { of, EMPTY } from 'rxjs';
 import { delayWhen } from 'rxjs/operators';
 import { TestScheduler } from 'rxjs/testing';
 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
@@ -106,6 +106,28 @@ describe('delayWhen operator', () => {
  expectObservable(result).toBe(expected);
  expectSubscriptions(e1.subscriptions).toBe(subs);
  expectSubscriptions(selector.subscriptions).toBe(selectorSubs);
+ });
+
+ it('should emit if the selector completes synchronously', () => {
+ const e1 = hot('a--|');
+ const expected = 'a--|';
+ const subs = '^ !';
+
+ const result = e1.pipe(delayWhen((x: any) => EMPTY));
+
+ expectObservable(result).toBe(expected);
+ expectSubscriptions(e1.subscriptions).toBe(subs);
+ });
+
+ it('should emit if the source completes synchronously and the selector completes synchronously', () => {
+ const e1 = hot('(a|)');
+ const expected = '(a|)';
+ const subs = '(^!)';
+
+ const result = e1.pipe(delayWhen((x: any) => EMPTY));
+
+ expectObservable(result).toBe(expected);
+ expectSubscriptions(e1.subscriptions).toBe(subs);
  });
 
  it('should not emit if selector never emits', () => {

src/internal/operators/delayWhen.ts

@@ -7,6 +7,12 @@ import { InnerSubscriber } from '../InnerSubscriber';
 import { subscribeToResult } from '../util/subscribeToResult';
 import { MonoTypeOperatorFunction, TeardownLogic } from '../types';
 
+/* tslint:disable:max-line-length */
+/** @deprecated In future versions, empty notifiers will no longer re-emit the source value on the output observable. */
+export function delayWhen<T>(delayDurationSelector: (value: T) => Observable<never>, subscriptionDelay?: Observable<any>): MonoTypeOperatorFunction<T>;
+export function delayWhen<T>(delayDurationSelector: (value: T) => Observable<any>, subscriptionDelay?: Observable<any>): MonoTypeOperatorFunction<T>;
+/* tslint:disable:max-line-length */
+
 /**
  * Delays the emission of items from the source Observable by a given time span
  * determined by the emissions of another Observable.
@@ -22,6 +28,8 @@ import { MonoTypeOperatorFunction, TeardownLogic } from '../types';
  * argument, and should return an Observable, called the "duration" Observable.
  * The source value is emitted on the output Observable only when the duration
  * Observable emits a value or completes.
+ * The completion of the notifier triggering the emission of the source value
+ * is deprecated behavior and will be removed in future versions.
  *
  * Optionally, `delayWhen` takes a second argument, `subscriptionDelay`, which
  * is an Observable. When `subscriptionDelay` emits its first value or
@@ -79,7 +87,6 @@ class DelayWhenOperator<T> implements Operator<T, T> {
 class DelayWhenSubscriber<T, R> extends OuterSubscriber<T, R> {
  private completed: boolean = false;
  private delayNotifierSubscriptions: Array<Subscription> = [];
- private values: Array<T> = [];
 
  constructor(destination: Subscriber<T>,
  private delayDurationSelector: (value: T) => Observable<any>) {
@@ -126,15 +133,11 @@ class DelayWhenSubscriber<T, R> extends OuterSubscriber<T, R> {
  subscription.unsubscribe();
 
  const subscriptionIdx = this.delayNotifierSubscriptions.indexOf(subscription);
- let value: T = null;
-
  if (subscriptionIdx !== -1) {
- value = this.values[subscriptionIdx];
  this.delayNotifierSubscriptions.splice(subscriptionIdx, 1);
- this.values.splice(subscriptionIdx, 1);
  }
 
- return value;
+ return subscription.outerValue;
  }
 
  private tryDelay(delayNotifier: Observable<any>, value: T): void {
@@ -144,8 +147,6 @@ class DelayWhenSubscriber<T, R> extends OuterSubscriber<T, R> {
  this.add(notifierSubscription);
  this.delayNotifierSubscriptions.push(notifierSubscription);
  }
-
- this.values.push(value);
  }
 
  private tryComplete(): void {