feat(distinct): distinct's flush supports ObservableInput

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

Hide file tree

Showing 3 changed files with 8 additions and 8 deletions.
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,8 @@
-import { Observable } from '../Observable';
-import { MonoTypeOperatorFunction } from '../types';
+import { MonoTypeOperatorFunction, ObservableInput } from '../types';
 import { operate } from '../util/lift';
 import { createOperatorSubscriber } from './OperatorSubscriber';
 import { noop } from '../util/noop';
+import { innerFrom } from '../observable/innerFrom';
 
 /**
  * Returns an Observable that emits all items emitted by the source Observable that are distinct by comparison from previous items.
@@ -56,12 +56,12 @@ import { noop } from '../util/noop';
  * @see {@link distinctUntilChanged}
  * @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 keySelector Optional `function` to select which value you want to check as distinct.
+ * @param 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 +74,6 @@ export function distinct<T, K>(keySelector?: (value: T) => K, flushes?: Observab
       })
     );
 
-    flushes?.subscribe(createOperatorSubscriber(subscriber, () => distinctKeys.clear(), noop));
+    flushes && innerFrom(flushes).subscribe(createOperatorSubscriber(subscriber, () => distinctKeys.clear(), noop));
   });
 }