feat(repeatWhen): added ObservableInput support for notifier

Author: jeremymwells

api_guard/dist/types/index.d.ts

@@ -577,7 +577,7 @@ export declare function refCount<T>(): MonoTypeOperatorFunction<T>;
 
 export declare function repeat<T>(countOrConfig?: number | RepeatConfig): MonoTypeOperatorFunction<T>;
 
-export declare function repeatWhen<T>(notifier: (notifications: Observable<void>) => Observable<any>): MonoTypeOperatorFunction<T>;
+export declare function repeatWhen<T>(notifier: (notifications: Observable<void>) => ObservableInput<any>): MonoTypeOperatorFunction<T>;
 
 export declare class ReplaySubject<T> extends Subject<T> {
     constructor(_bufferSize?: number, _windowTime?: number, _timestampProvider?: TimestampProvider);

api_guard/dist/types/operators/index.d.ts

@@ -231,7 +231,7 @@ export declare function refCount<T>(): MonoTypeOperatorFunction<T>;
 
 export declare function repeat<T>(countOrConfig?: number | RepeatConfig): MonoTypeOperatorFunction<T>;
 
-export declare function repeatWhen<T>(notifier: (notifications: Observable<void>) => Observable<any>): MonoTypeOperatorFunction<T>;
+export declare function repeatWhen<T>(notifier: (notifications: Observable<void>) => ObservableInput<any>): MonoTypeOperatorFunction<T>;
 
 export declare function retry<T>(count?: number): MonoTypeOperatorFunction<T>;
 export declare function retry<T>(config: RetryConfig): MonoTypeOperatorFunction<T>;

src/internal/operators/repeatWhen.ts

@@ -1,8 +1,9 @@
 import { Observable } from '../Observable';
+import { innerFrom } from '../observable/innerFrom';
 import { Subject } from '../Subject';
 import { Subscription } from '../Subscription';
 
-import { MonoTypeOperatorFunction } from '../types';
+import { MonoTypeOperatorFunction, ObservableInput } from '../types';
 import { operate } from '../util/lift';
 import { createOperatorSubscriber } from './OperatorSubscriber';
 
@@ -33,13 +34,13 @@ import { createOperatorSubscriber } from './OperatorSubscriber';
  * @see {@link retry}
  * @see {@link retryWhen}
  *
- * @param {function(notifications: Observable): Observable} notifier - Receives an Observable of notifications with
+ * @param notifier function that receives an Observable of notifications with
  * which a user can `complete` or `error`, aborting the repetition.
- * @return A function that returns an Observable that mirrors the source
+ * @return A function that returns an `ObservableInput` that mirrors the source
  * Observable with the exception of a `complete`.
  * @deprecated Will be removed in v9 or v10. Use {@link repeat}'s `delay` option instead.
  */
-export function repeatWhen<T>(notifier: (notifications: Observable<void>) => Observable<any>): MonoTypeOperatorFunction<T> {
+export function repeatWhen<T>(notifier: (notifications: Observable<void>) => ObservableInput<any>): MonoTypeOperatorFunction<T> {
   return operate((source, subscriber) => {
     let innerSub: Subscription | null;
     let syncResub = false;
@@ -61,7 +62,7 @@ export function repeatWhen<T>(notifier: (notifications: Observable<void>) => Obs
 
         // If the call to `notifier` throws, it will be caught by the OperatorSubscriber
         // In the main subscription -- in `subscribeForRepeatWhen`.
-        notifier(completions$).subscribe(
+        innerFrom(notifier(completions$)).subscribe(
           createOperatorSubscriber(
             subscriber,
             () => {