feat(distinct): distinct's flush supports ObservableInput

ReactiveX · Oct 9, 2022 · c7cd5e5 · c7cd5e5
1 parent afac3d5
commit c7cd5e5
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 6 deletions.
diff --git a/.travis.yml b/.travis.yml
diff --git a/api_guard/dist/types/index.d.ts b/api_guard/dist/types/index.d.ts
@@ -160,7 +160,7 @@ export declare function delayWhen<T>(delayDurationSelector: (value: T, index: nu
 
 export declare function dematerialize<N extends ObservableNotification<any>>(): OperatorFunction<N, ValueFromNotification<N>>;
 
-export declare function distinct<T, K>(keySelector?: (value: T) => K, flushes?: Observable<any>): MonoTypeOperatorFunction<T>;
+export declare function distinct<T, K>(keySelector?: (value: T) => K, flushes?: ObservableInput<any>): MonoTypeOperatorFunction<T>;
 
 export declare function distinctUntilChanged<T>(comparator?: (previous: T, current: T) => boolean): MonoTypeOperatorFunction<T>;
 export declare function distinctUntilChanged<T, K>(comparator: (previous: K, current: K) => boolean, keySelector: (value: T) => K): MonoTypeOperatorFunction<T>;

diff --git a/api_guard/dist/types/operators/index.d.ts b/api_guard/dist/types/operators/index.d.ts
@@ -72,7 +72,7 @@ export declare function delayWhen<T>(delayDurationSelector: (value: T, index: nu
 
 export declare function dematerialize<N extends ObservableNotification<any>>(): OperatorFunction<N, ValueFromNotification<N>>;
 
-export declare function distinct<T, K>(keySelector?: (value: T) => K, flushes?: Observable<any>): MonoTypeOperatorFunction<T>;
+export declare function distinct<T, K>(keySelector?: (value: T) => K, flushes?: ObservableInput<any>): MonoTypeOperatorFunction<T>;
 
 export declare function distinctUntilChanged<T>(comparator?: (previous: T, current: T) => boolean): MonoTypeOperatorFunction<T>;
 export declare function distinctUntilChanged<T, K>(comparator: (previous: K, current: K) => boolean, keySelector: (value: T) => K): MonoTypeOperatorFunction<T>;

diff --git a/src/internal/operators/distinct.ts b/src/internal/operators/distinct.ts
@@ -1,8 +1,12 @@
 import { Observable } from '../Observable';
-import { MonoTypeOperatorFunction } from '../types';
+import { InteropObservable, MonoTypeOperatorFunction, ObservableInput } from '../types';
 import { operate } from '../util/lift';
 import { createOperatorSubscriber } from './OperatorSubscriber';
 import { noop } from '../util/noop';
+import { observable as Symbol_observable } from '../symbol/observable';
+import { of } from '../observable/of';
+import { isInteropObservable } from '../util/isInteropObservable';
+import { fromInteropObservable } from '../observable/innerFrom';
 
 /**
  * Returns an Observable that emits all items emitted by the source Observable that are distinct by comparison from previous items.
@@ -57,11 +61,11 @@ import { noop } from '../util/noop';
  * @see {@link distinctUntilKeyChanged}
  *
  * @param {function} [keySelector] Optional function to select which value you want to check as distinct.
- * @param {Observable} [flushes] Optional Observable for flushing the internal HashSet of the operator.
+ * @param {ObservableInput} [flushes] Optional ObservableInput for flushing the internal HashSet of the operator.
  * @return A function that returns an Observable that emits items from the
  * source Observable with distinct values.
  */
-export function distinct<T, K>(keySelector?: (value: T) => K, flushes?: Observable<any>): MonoTypeOperatorFunction<T> {
+export function distinct<T, K>(keySelector?: (value: T) => K, flushes?: ObservableInput<any>): MonoTypeOperatorFunction<T> {
   return operate((source, subscriber) => {
     const distinctKeys = new Set();
     source.subscribe(
@@ -74,6 +78,11 @@ export function distinct<T, K>(keySelector?: (value: T) => K, flushes?: Observab
       })
     );
 
-    flushes?.subscribe(createOperatorSubscriber(subscriber, () => distinctKeys.clear(), noop));
+    let flush$ = flushes as InteropObservable<any> | ObservableInput<any> | undefined | any;
+    if (flushes && isInteropObservable(flushes)) {
+      flush$ = fromInteropObservable(flushes);
+    }
+
+    flush$?.subscribe(createOperatorSubscriber(subscriber, () => distinctKeys.clear(), noop));
   });
 }