feat(distinct): distinct's flush supports ObservableInput

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>;

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>;

spec-dtslint/operators/distinct-spec.ts

@@ -1,5 +1,7 @@
 import { of } from 'rxjs';
+import { asInteropObservable } from '../../spec/helpers/interop-helper';
 import { distinct } from 'rxjs/operators';
+import { ReadableStreamLike } from '../../src/internal/types';
 
 it('should infer correctly', () => {
   const o = of(1, 2, 3).pipe(distinct()); // $ExpectType Observable<number>
@@ -10,10 +12,79 @@ it('should accept a keySelector', () => {
   const o = of({ name: 'Tim' } as Person).pipe(distinct(person => person.name)); // $ExpectType Observable<Person>
 });
 
-it('should accept flushes', () => {
+it('should accept observable flush', () => {
   const o = of(1, 2, 3).pipe(distinct(n => n, of('t', 'i', 'm'))); // $ExpectType Observable<number>
 });
 
+it('should accept interop observable flush', () => {
+  of(1, 2, 3).pipe(distinct(n => n, asInteropObservable(of('t', 'i', 'm')))); // $ExpectType Observable<number>
+});
+
+it('should accept array-like flush', () => {
+  of(1, 2, 3).pipe(distinct(n => n, [1,2,3])); // $ExpectType Observable<number>
+});
+
+it('should accept promise flush', () => {
+  of(1, 2, 3).pipe(distinct(n => n, Promise.resolve())); // $ExpectType Observable<number>
+});
+
+it('should accept async iterable flush', () => {
+  const asyncRange = {
+    from: 1,
+    to: 2,
+    [Symbol.asyncIterator]() {
+      return {
+        current: this.from,
+        last: this.to,
+        async next() {
+          await Promise.resolve();
+          const done = (this.current > this.last);
+          return {
+            done,
+            value: done ? this.current++ : undefined
+          };
+        }
+      };
+    }
+  };
+  of(1, 2, 3).pipe(distinct(n => n, asyncRange)); // $ExpectType Observable<number>
+});
+
+it('should accept iterable flush', () => {
+  const syncRange = {
+    from: 1,
+    to: 2,
+    [Symbol.iterator]() {
+      return {
+        current: this.from,
+        last: this.to,
+        next() {
+          const done = (this.current > this.last);
+          return {
+            done,
+            value: done ? this.current++ : undefined
+          };
+        }
+      };
+    }
+  };
+  of(1, 2, 3).pipe(distinct(n => n, syncRange)); // $ExpectType Observable<number>
+});
+
+it('should accept readable stream flush', () => {
+  const readable: ReadableStreamLike<string> = new ReadableStream<string>({
+    pull(controller) {
+      controller.enqueue('x');
+      controller.close();
+    },
+  });
+  of(1, 2, 3).pipe(distinct(n => n, readable)); // $ExpectType Observable<number>
+});
+
+it('should error with unsupported flush', () => {
+  of(1, 2, 3).pipe(distinct(n => n, {})); // $ExpectError
+});
+
 it('should enforce types', () => {
   const o = of(1, 2, 3).pipe(distinct('F00D')); // $ExpectError
 });

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));
   });
 }