Skip to content

refactor(merge): support N-args with concurrency and scheduler #5877

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
benlesh merged 2 commits into from
Nov 17, 2020
Merged

refactor(merge): support N-args with concurrency and scheduler #5877

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e13ff11
refactor(merge): N-args with scheduler
cartant Nov 8, 2020
5c5da02
chore: update api_guardian
cartant Nov 8, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 4 additions & 28 deletions api_guard/dist/types/index.d.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -210,34 +210,10 @@ export declare function isObservable<T>(obj: any): obj is Observable<T>;

export declare function lastValueFrom<T>(source: Observable<T>): Promise<T>;

export declare function merge<T>(v1: ObservableInput<T>, scheduler: SchedulerLike): Observable<T>;
export declare function merge<T>(v1: ObservableInput<T>, concurrent: number, scheduler: SchedulerLike): Observable<T>;
export declare function merge<T, T2>(v1: ObservableInput<T>, v2: ObservableInput<T2>, scheduler: SchedulerLike): Observable<T | T2>;
export declare function merge<T, T2>(v1: ObservableInput<T>, v2: ObservableInput<T2>, concurrent: number, scheduler: SchedulerLike): Observable<T | T2>;
export declare function merge<T, T2, T3>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, scheduler: SchedulerLike): Observable<T | T2 | T3>;
export declare function merge<T, T2, T3>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, concurrent: number, scheduler: SchedulerLike): Observable<T | T2 | T3>;
export declare function merge<T, T2, T3, T4>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, scheduler: SchedulerLike): Observable<T | T2 | T3 | T4>;
export declare function merge<T, T2, T3, T4>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, concurrent: number, scheduler: SchedulerLike): Observable<T | T2 | T3 | T4>;
export declare function merge<T, T2, T3, T4, T5>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, v5: ObservableInput<T5>, scheduler: SchedulerLike): Observable<T | T2 | T3 | T4 | T5>;
export declare function merge<T, T2, T3, T4, T5>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, v5: ObservableInput<T5>, concurrent: number, scheduler: SchedulerLike): Observable<T | T2 | T3 | T4 | T5>;
export declare function merge<T, T2, T3, T4, T5, T6>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, v5: ObservableInput<T5>, v6: ObservableInput<T6>, scheduler: SchedulerLike): Observable<T | T2 | T3 | T4 | T5 | T6>;
export declare function merge<T, T2, T3, T4, T5, T6>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, v5: ObservableInput<T5>, v6: ObservableInput<T6>, concurrent: number, scheduler: SchedulerLike): Observable<T | T2 | T3 | T4 | T5 | T6>;
export declare function merge<T>(v1: ObservableInput<T>): Observable<T>;
export declare function merge<T>(v1: ObservableInput<T>, concurrent: number): Observable<T>;
export declare function merge<T, T2>(v1: ObservableInput<T>, v2: ObservableInput<T2>): Observable<T | T2>;
export declare function merge<T, T2>(v1: ObservableInput<T>, v2: ObservableInput<T2>, concurrent: number): Observable<T | T2>;
export declare function merge<T, T2, T3>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>): Observable<T | T2 | T3>;
export declare function merge<T, T2, T3>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, concurrent: number): Observable<T | T2 | T3>;
export declare function merge<T, T2, T3, T4>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>): Observable<T | T2 | T3 | T4>;
export declare function merge<T, T2, T3, T4>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, concurrent: number): Observable<T | T2 | T3 | T4>;
export declare function merge<T, T2, T3, T4, T5>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, v5: ObservableInput<T5>): Observable<T | T2 | T3 | T4 | T5>;
export declare function merge<T, T2, T3, T4, T5>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, v5: ObservableInput<T5>, concurrent: number): Observable<T | T2 | T3 | T4 | T5>;
export declare function merge<T, T2, T3, T4, T5, T6>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, v5: ObservableInput<T5>, v6: ObservableInput<T6>): Observable<T | T2 | T3 | T4 | T5 | T6>;
export declare function merge<T, T2, T3, T4, T5, T6>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, v5: ObservableInput<T5>, v6: ObservableInput<T6>, concurrent: number): Observable<T | T2 | T3 | T4 | T5 | T6>;
export declare function merge<T>(...observables: (ObservableInput<T> | number)[]): Observable<T>;
export declare function merge<T>(...observables: (ObservableInput<T> | SchedulerLike | number)[]): Observable<T>;
export declare function merge<T, R>(...observables: (ObservableInput<any> | number)[]): Observable<R>;
export declare function merge<T, R>(...observables: (ObservableInput<any> | SchedulerLike | number)[]): Observable<R>;
export declare function merge<A extends readonly unknown[]>(...args: [...ObservableInputTuple<A>]): Observable<A[number]>;
export declare function merge<A extends readonly unknown[]>(...args: [...ObservableInputTuple<A>, number?]): Observable<A[number]>;
export declare function merge<A extends readonly unknown[]>(...args: [...ObservableInputTuple<A>, SchedulerLike?]): Observable<A[number]>;
export declare function merge<A extends readonly unknown[]>(...args: [...ObservableInputTuple<A>, number?, SchedulerLike?]): Observable<A[number]>;

export interface MonoTypeOperatorFunction<T> extends OperatorFunction<T, T> {
}
Expand Down
165 changes: 11 additions & 154 deletions src/internal/observable/merge.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,163 +1,20 @@
/** @prettier */
import { Observable } from '../Observable';
import { ObservableInput, SchedulerLike } from '../types';
import { ObservableInput, ObservableInputTuple, SchedulerLike } from '../types';
import { mergeAll } from '../operators/mergeAll';
import { internalFromArray } from './fromArray';
import { argsOrArgArray } from '../util/argsOrArgArray';
import { innerFrom } from './from';
import { EMPTY } from './empty';
import { popNumber, popScheduler } from '../util/args';

/* tslint:disable:max-line-length */
export function merge<A extends readonly unknown[]>(...args: [...ObservableInputTuple<A>]): Observable<A[number]>;
export function merge<A extends readonly unknown[]>(...args: [...ObservableInputTuple<A>, number?]): Observable<A[number]>;
/** @deprecated The scheduler argument is deprecated, use scheduled and mergeAll. Details: https://rxjs.dev/deprecations/scheduler-argument */
export function merge<T>(v1: ObservableInput<T>, scheduler: SchedulerLike): Observable<T>;
export function merge<A extends readonly unknown[]>(...args: [...ObservableInputTuple<A>, SchedulerLike?]): Observable<A[number]>;
/** @deprecated The scheduler argument is deprecated, use scheduled and mergeAll. Details: https://rxjs.dev/deprecations/scheduler-argument */
export function merge<T>(v1: ObservableInput<T>, concurrent: number, scheduler: SchedulerLike): Observable<T>;
/** @deprecated The scheduler argument is deprecated, use scheduled and mergeAll. Details: https://rxjs.dev/deprecations/scheduler-argument */
export function merge<T, T2>(v1: ObservableInput<T>, v2: ObservableInput<T2>, scheduler: SchedulerLike): Observable<T | T2>;
/** @deprecated The scheduler argument is deprecated, use scheduled and mergeAll. Details: https://rxjs.dev/deprecations/scheduler-argument */
export function merge<T, T2>(
v1: ObservableInput<T>,
v2: ObservableInput<T2>,
concurrent: number,
scheduler: SchedulerLike
): Observable<T | T2>;
/** @deprecated The scheduler argument is deprecated, use scheduled and mergeAll. Details: https://rxjs.dev/deprecations/scheduler-argument */
export function merge<T, T2, T3>(
v1: ObservableInput<T>,
v2: ObservableInput<T2>,
v3: ObservableInput<T3>,
scheduler: SchedulerLike
): Observable<T | T2 | T3>;
/** @deprecated The scheduler argument is deprecated, use scheduled and mergeAll. Details: https://rxjs.dev/deprecations/scheduler-argument */
export function merge<T, T2, T3>(
v1: ObservableInput<T>,
v2: ObservableInput<T2>,
v3: ObservableInput<T3>,
concurrent: number,
scheduler: SchedulerLike
): Observable<T | T2 | T3>;
/** @deprecated The scheduler argument is deprecated, use scheduled and mergeAll. Details: https://rxjs.dev/deprecations/scheduler-argument */
export function merge<T, T2, T3, T4>(
v1: ObservableInput<T>,
v2: ObservableInput<T2>,
v3: ObservableInput<T3>,
v4: ObservableInput<T4>,
scheduler: SchedulerLike
): Observable<T | T2 | T3 | T4>;
/** @deprecated The scheduler argument is deprecated, use scheduled and mergeAll. Details: https://rxjs.dev/deprecations/scheduler-argument */
export function merge<T, T2, T3, T4>(
v1: ObservableInput<T>,
v2: ObservableInput<T2>,
v3: ObservableInput<T3>,
v4: ObservableInput<T4>,
concurrent: number,
scheduler: SchedulerLike
): Observable<T | T2 | T3 | T4>;
/** @deprecated The scheduler argument is deprecated, use scheduled and mergeAll. Details: https://rxjs.dev/deprecations/scheduler-argument */
export function merge<T, T2, T3, T4, T5>(
v1: ObservableInput<T>,
v2: ObservableInput<T2>,
v3: ObservableInput<T3>,
v4: ObservableInput<T4>,
v5: ObservableInput<T5>,
scheduler: SchedulerLike
): Observable<T | T2 | T3 | T4 | T5>;
/** @deprecated The scheduler argument is deprecated, use scheduled and mergeAll. Details: https://rxjs.dev/deprecations/scheduler-argument */
export function merge<T, T2, T3, T4, T5>(
v1: ObservableInput<T>,
v2: ObservableInput<T2>,
v3: ObservableInput<T3>,
v4: ObservableInput<T4>,
v5: ObservableInput<T5>,
concurrent: number,
scheduler: SchedulerLike
): Observable<T | T2 | T3 | T4 | T5>;
/** @deprecated The scheduler argument is deprecated, use scheduled and mergeAll. Details: https://rxjs.dev/deprecations/scheduler-argument */
export function merge<T, T2, T3, T4, T5, T6>(
v1: ObservableInput<T>,
v2: ObservableInput<T2>,
v3: ObservableInput<T3>,
v4: ObservableInput<T4>,
v5: ObservableInput<T5>,
v6: ObservableInput<T6>,
scheduler: SchedulerLike
): Observable<T | T2 | T3 | T4 | T5 | T6>;
/** @deprecated The scheduler argument is deprecated, use scheduled and mergeAll. Details: https://rxjs.dev/deprecations/scheduler-argument */
export function merge<T, T2, T3, T4, T5, T6>(
v1: ObservableInput<T>,
v2: ObservableInput<T2>,
v3: ObservableInput<T3>,
v4: ObservableInput<T4>,
v5: ObservableInput<T5>,
v6: ObservableInput<T6>,
concurrent: number,
scheduler: SchedulerLike
): Observable<T | T2 | T3 | T4 | T5 | T6>;
export function merge<A extends readonly unknown[]>(...args: [...ObservableInputTuple<A>, number?, SchedulerLike?]): Observable<A[number]>;

export function merge<T>(v1: ObservableInput<T>): Observable<T>;
export function merge<T>(v1: ObservableInput<T>, concurrent: number): Observable<T>;
export function merge<T, T2>(v1: ObservableInput<T>, v2: ObservableInput<T2>): Observable<T | T2>;
export function merge<T, T2>(v1: ObservableInput<T>, v2: ObservableInput<T2>, concurrent: number): Observable<T | T2>;
export function merge<T, T2, T3>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>): Observable<T | T2 | T3>;
export function merge<T, T2, T3>(
v1: ObservableInput<T>,
v2: ObservableInput<T2>,
v3: ObservableInput<T3>,
concurrent: number
): Observable<T | T2 | T3>;
export function merge<T, T2, T3, T4>(
v1: ObservableInput<T>,
v2: ObservableInput<T2>,
v3: ObservableInput<T3>,
v4: ObservableInput<T4>
): Observable<T | T2 | T3 | T4>;
export function merge<T, T2, T3, T4>(
v1: ObservableInput<T>,
v2: ObservableInput<T2>,
v3: ObservableInput<T3>,
v4: ObservableInput<T4>,
concurrent: number
): Observable<T | T2 | T3 | T4>;
export function merge<T, T2, T3, T4, T5>(
v1: ObservableInput<T>,
v2: ObservableInput<T2>,
v3: ObservableInput<T3>,
v4: ObservableInput<T4>,
v5: ObservableInput<T5>
): Observable<T | T2 | T3 | T4 | T5>;
export function merge<T, T2, T3, T4, T5>(
v1: ObservableInput<T>,
v2: ObservableInput<T2>,
v3: ObservableInput<T3>,
v4: ObservableInput<T4>,
v5: ObservableInput<T5>,
concurrent: number
): Observable<T | T2 | T3 | T4 | T5>;
export function merge<T, T2, T3, T4, T5, T6>(
v1: ObservableInput<T>,
v2: ObservableInput<T2>,
v3: ObservableInput<T3>,
v4: ObservableInput<T4>,
v5: ObservableInput<T5>,
v6: ObservableInput<T6>
): Observable<T | T2 | T3 | T4 | T5 | T6>;
export function merge<T, T2, T3, T4, T5, T6>(
v1: ObservableInput<T>,
v2: ObservableInput<T2>,
v3: ObservableInput<T3>,
v4: ObservableInput<T4>,
v5: ObservableInput<T5>,
v6: ObservableInput<T6>,
concurrent: number
): Observable<T | T2 | T3 | T4 | T5 | T6>;
export function merge<T>(...observables: (ObservableInput<T> | number)[]): Observable<T>;
/** @deprecated use {@link scheduled} and {@link mergeAll} (e.g. `scheduled([ob1, ob2, ob3], scheduler).pipe(mergeAll())*/
export function merge<T>(...observables: (ObservableInput<T> | SchedulerLike | number)[]): Observable<T>;
export function merge<T, R>(...observables: (ObservableInput<any> | number)[]): Observable<R>;
/** @deprecated use {@link scheduled} and {@link mergeAll} (e.g. `scheduled([ob1, ob2, ob3], scheduler).pipe(mergeAll())*/
export function merge<T, R>(...observables: (ObservableInput<any> | SchedulerLike | number)[]): Observable<R>;
/* tslint:enable:max-line-length */
/**
* Creates an output Observable which concurrently emits all values from every
* given input Observable.
Expand Down Expand Up @@ -225,16 +82,16 @@ export function merge<T, R>(...observables: (ObservableInput<any> | SchedulerLik
* @return {Observable} an Observable that emits items that are the result of
* every input Observable.
*/
export function merge(...args: (ObservableInput<any> | SchedulerLike | number)[]): Observable<unknown> {
export function merge(...args: (ObservableInput<unknown> | number | SchedulerLike)[]): Observable<unknown> {
const scheduler = popScheduler(args);
const concurrent = popNumber(args, Infinity);
args = argsOrArgArray(args);
return !args.length
const sources = argsOrArgArray(args) as ObservableInput<unknown>[];
return !sources.length
? // No source provided
EMPTY
: args.length === 1
: sources.length === 1
? // One source? Just return it.
innerFrom(args[0] as ObservableInput<any>)
innerFrom(sources[0])
: // Merge all sources
mergeAll(concurrent)(internalFromArray(args as ObservableInput<any>[], scheduler));
mergeAll(concurrent)(internalFromArray(sources, scheduler));
}