fix(expand): operator function signature

spec-dtslint/operators/expand-spec.ts

@@ -1,51 +1,65 @@
-import { of, asyncScheduler } from 'rxjs';
-import { expand } from 'rxjs/operators';
-
-it('should infer correctly', () => {
-  const o = of(1, 2, 3).pipe(expand(value => of(value))); // $ExpectType Observable<number>
-  const p = of(1, 2, 3).pipe(expand(value => [value])); // $ExpectType Observable<number>
-  const q = of(1, 2, 3).pipe(expand(value => Promise.resolve(value))); // $ExpectType Observable<number>
-});
-
-it('should infer correctly with a different type as the source', () => {
-  const o = of(1, 2, 3).pipe(expand(value => of('foo'))); // $ExpectType Observable<string>
-  const p = of(1, 2, 3).pipe(expand(value => ['foo'])); // $ExpectType Observable<string>
-  const q = of(1, 2, 3).pipe(expand(value => Promise.resolve('foo'))); // $ExpectType Observable<string>
-});
-
-it('should support a project function with index', () => {
-  const o = of(1, 2, 3).pipe(expand((value, index) => of(index))); // $ExpectType Observable<number>
-});
-
-it('should support concurrent parameter', () => {
-  const o = of(1, 2, 3).pipe(expand(value => of(1), 47)); // $ExpectType Observable<number>
-});
-
-it('should support a scheduler', () => {
-  const o = of(1, 2, 3).pipe(expand(value => of(1), 47, asyncScheduler)); // $ExpectType Observable<number>
-});
-
-it('should enforce types', () => {
-  const o = of(1, 2, 3).pipe(expand()); // $ExpectError
-});
-
-it('should enforce project types', () => {
-  const o = of(1, 2, 3).pipe(expand((value: string, index) => of(1))); // $ExpectError
-  const p = of(1, 2, 3).pipe(expand((value, index: string) => of(1))); // $ExpectError
-});
-
-it('should enforce project return type', () => {
-  const o = of(1, 2, 3).pipe(expand(value => 1)); // $ExpectError
-});
-
-it('should enforce concurrent type', () => {
-  const o = of(1, 2, 3).pipe(expand(value => of(1), 'foo')); // $ExpectError
-});
-
-it('should enforce scheduler type', () => {
-  const o = of(1, 2, 3).pipe(expand(value => of(1), 47, 'foo')); // $ExpectError
-});
-
-it('should support union types', () => {
-  const o = of(1).pipe(expand(x => typeof x === 'string' ? of(123) : of('test'))); // $ExpectType Observable<string | number>
-});
+import { of, asyncScheduler } from 'rxjs';
+import { expand } from 'rxjs/operators';
+
+it('should infer correctly', () => {
+  const o = of(1, 2, 3).pipe(expand(value => of(value))); // $ExpectType Observable<number>
+  const p = of(1, 2, 3).pipe(expand(value => [value])); // $ExpectType Observable<number>
+  const q = of(1, 2, 3).pipe(expand(value => Promise.resolve(value))); // $ExpectType Observable<number>
+});
+
+it('should infer correctly specifying value argument type', () => {
+  const o = of(1).pipe(expand((value: number | string) => of(value.toString()))); // $ExpectType Observable<string | number>
+});
+
+it('should infer correctly with specifying different input/output types', () => {
+  const o = of(1).pipe(expand<number, string>((value) => of(value.toString()))); // $ExpectType Observable<string | number>
+});
+
+it('should enforce project output type to be assignable with its generic', () => {
+  const o = of(1).pipe(expand<number, string>((value) => of(value))); // $ExpectError
+});
+
+it('should enforce project input type to be assignable with upstream', () => {
+  const o = of(1).pipe(expand((value: string) => of(value))); // $ExpectError
+});
+
+it('should enforce project input/output types compatibility by default', () => {
+  const o = of(1).pipe(expand((value) => of(value.toString()))); // $ExpectError
+});
+
+it('should support a project function with index', () => {
+  const o = of(1, 2, 3).pipe(expand((value, index) => of(index))); // $ExpectType Observable<number>
+});
+
+it('should support concurrent parameter', () => {
+  const o = of(1, 2, 3).pipe(expand(value => of(1), 47)); // $ExpectType Observable<number>
+});
+
+it('should support a scheduler', () => {
+  const o = of(1, 2, 3).pipe(expand(value => of(1), 47, asyncScheduler)); // $ExpectType Observable<number>
+});
+
+it('should enforce types', () => {
+  const o = of(1, 2, 3).pipe(expand()); // $ExpectError
+});
+
+it('should enforce project types', () => {
+  const o = of(1, 2, 3).pipe(expand((value: string, index) => of(1))); // $ExpectError
+  const p = of(1, 2, 3).pipe(expand((value, index: string) => of(1))); // $ExpectError
+});
+
+it('should enforce project return type', () => {
+  const o = of(1, 2, 3).pipe(expand(value => 1)); // $ExpectError
+});
+
+it('should enforce concurrent type', () => {
+  const o = of(1, 2, 3).pipe(expand(value => of(1), 'foo')); // $ExpectError
+});
+
+it('should enforce scheduler type', () => {
+  const o = of(1, 2, 3).pipe(expand(value => of(1), 47, 'foo')); // $ExpectError
+});
+
+it('should support union types', () => {
+  const o = of(1).pipe(expand<string | number>(x => typeof x === 'string' ? of(123) : of('test'))); // $ExpectType Observable<string | number>
+});

spec/operators/expand-spec.ts

@@ -511,7 +511,7 @@ describe('expand', () => {
 
     synchronousObservable
       .pipe(
-        expand(() => EMPTY),
+        expand((_) => EMPTY),
         take(3)
       )
       .subscribe(() => {

src/internal/operators/expand.ts

@@ -1,23 +1,23 @@
-import { OperatorFunction, ObservableInput, ObservedValueOf, SchedulerLike } from '../types';
+import { ObservableInput, SchedulerLike, OperatorFunction } from '../types';
 import { Observable } from '../Observable';
 import { mergeInternals } from './mergeInternals';
 
 /* tslint:disable:max-line-length */
-export function expand<T, O extends ObservableInput<unknown>>(
-  project: (value: T, index: number) => O,
+export function expand<I, O = I>(
+  project: (value: I | O, index: number) => ObservableInput<O>,
   concurrent?: number,
   scheduler?: SchedulerLike
-): OperatorFunction<T, ObservedValueOf<O>>;
+): OperatorFunction<I, I | O>;
 /**
  * @deprecated The `scheduler` parameter will be removed in v8. If you need to schedule the inner subscription,
  * use `subscribeOn` within the projection function: `expand((value) => fn(value).pipe(subscribeOn(scheduler)))`.
  * Details: Details: https://rxjs.dev/deprecations/scheduler-argument
  */
-export function expand<T, O extends ObservableInput<unknown>>(
-  project: (value: T, index: number) => O,
+export function expand<I, O = I>(
+  project: (value: I | O, index: number) => ObservableInput<O>,
   concurrent: number | undefined,
   scheduler: SchedulerLike
-): OperatorFunction<T, ObservedValueOf<O>>;
+): OperatorFunction<I, I | O>;
 /* tslint:enable:max-line-length */
 
 /**
@@ -70,11 +70,11 @@ export function expand<T, O extends ObservableInput<unknown>>(
  * the output Observable and merging the results of the Observables obtained
  * from this transformation.
  */
-export function expand<T, O extends ObservableInput<unknown>>(
-  project: (value: T, index: number) => O,
+export function expand<I, O = I>(
+  project: (value: I | O, index: number) => ObservableInput<O>,
   concurrent = Infinity,
   scheduler?: SchedulerLike
-): OperatorFunction<T, ObservedValueOf<O>> {
+): OperatorFunction<I, I | O> {
   concurrent = (concurrent || 0) < 1 ? Infinity : concurrent;
   return (source) =>
     new Observable((subscriber) =>
@@ -83,8 +83,7 @@ export function expand<T, O extends ObservableInput<unknown>>(
         source,
         subscriber,
 
-        // HACK: Cast because TypeScript seems to get confused here.
-        project as (value: T, index: number) => ObservableInput<ObservedValueOf<O>>,
+        project,
         concurrent,
 
         // onBeforeNext