Skip to content
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.

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

feat(onErrorResumeNextWith): renamed onErrorResumeNext and exported from the top level #6755

Merged
merged 2 commits into from Dec 3, 2022
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
3 changes: 3 additions & 0 deletions api_guard/dist/types/index.d.ts
Expand Up @@ -500,6 +500,9 @@ export declare function of<A extends readonly unknown[]>(...values: A): Observab
export declare function onErrorResumeNext<A extends readonly unknown[]>(sources: [...ObservableInputTuple<A>]): Observable<A[number]>;
export declare function onErrorResumeNext<A extends readonly unknown[]>(...sources: [...ObservableInputTuple<A>]): Observable<A[number]>;

export declare function onErrorResumeNextWith<T, A extends readonly unknown[]>(sources: [...ObservableInputTuple<A>]): OperatorFunction<T, T | A[number]>;
export declare function onErrorResumeNextWith<T, A extends readonly unknown[]>(...sources: [...ObservableInputTuple<A>]): OperatorFunction<T, T | A[number]>;

export interface Operator<T, R> {
call(subscriber: Subscriber<R>, source: any): TeardownLogic;
}
Expand Down
3 changes: 1 addition & 2 deletions api_guard/dist/types/operators/index.d.ts
Expand Up @@ -191,8 +191,7 @@ export declare function multicast<T, O extends ObservableInput<any>>(subjectFact

export declare function observeOn<T>(scheduler: SchedulerLike, delay?: number): MonoTypeOperatorFunction<T>;

export declare function onErrorResumeNext<T, A extends readonly unknown[]>(sources: [...ObservableInputTuple<A>]): OperatorFunction<T, T | A[number]>;
export declare function onErrorResumeNext<T, A extends readonly unknown[]>(...sources: [...ObservableInputTuple<A>]): OperatorFunction<T, T | A[number]>;
export declare const onErrorResumeNext: typeof onErrorResumeNextWith;

export declare function pairwise<T>(): OperatorFunction<T, [T, T]>;

Expand Down
5 changes: 3 additions & 2 deletions docs_app/content/guide/importing.md
Expand Up @@ -56,11 +56,12 @@ ending with `With`). Those are:
| [`combineLatest`](/api/operators/combineLatest) | {@link combineLatest} | {@link combineLatestWith} |
| [`concat`](/api/operators/concat) | {@link concat} | {@link concatWith} |
| [`merge`](/api/operators/merge) | {@link merge} | {@link mergeWith} |
| [`onErrorResumeNext`](/api/operators/onErrorResumeNext) | {@link onErrorResumeNext} | - |
| [`partition`](/api/operators/partition) | {@link partition} | - |
| [`onErrorResumeNext`](/api/operators/onErrorResumeNext) | {@link onErrorResumeNext} | {@link onErrorResumeNextWith} |
| [`race`](/api/operators/race) | {@link race} | {@link raceWith} |
| [`zip`](/api/operators/zip) | {@link zip} | {@link zipWith} |

`partition`, the operator, is a special case, as it is deprecated and you should be using the `partition` creation function exported from `'rxjs'` instead.

For example, the old and deprecated way of using [`merge`](/api/operators/merge) from `'rxjs/operators'`
is:

Expand Down
1 change: 1 addition & 0 deletions src/index.ts
Expand Up @@ -155,6 +155,7 @@ export { mergeWith } from './internal/operators/mergeWith';
export { min } from './internal/operators/min';
export { multicast } from './internal/operators/multicast';
export { observeOn } from './internal/operators/observeOn';
export { onErrorResumeNextWith } from './internal/operators/onErrorResumeNextWith';
export { pairwise } from './internal/operators/pairwise';
export { pluck } from './internal/operators/pluck';
export { publish } from './internal/operators/publish';
Expand Down
28 changes: 25 additions & 3 deletions src/internal/observable/onErrorResumeNext.ts
@@ -1,8 +1,9 @@
import { Observable } from '../Observable';
import { ObservableInputTuple } from '../types';
import { EMPTY } from './empty';
import { onErrorResumeNext as onErrorResumeNextWith } from '../operators/onErrorResumeNext';
import { argsOrArgArray } from '../util/argsOrArgArray';
import { OperatorSubscriber } from '../operators/OperatorSubscriber';
import { noop } from '../util/noop';
import { innerFrom } from './innerFrom';

/* tslint:disable:max-line-length */
export function onErrorResumeNext<A extends readonly unknown[]>(sources: [...ObservableInputTuple<A>]): Observable<A[number]>;
Expand Down Expand Up @@ -75,5 +76,26 @@ export function onErrorResumeNext<A extends readonly unknown[]>(...sources: [...
export function onErrorResumeNext<A extends readonly unknown[]>(
...sources: [[...ObservableInputTuple<A>]] | [...ObservableInputTuple<A>]
): Observable<A[number]> {
return onErrorResumeNextWith(argsOrArgArray(sources))(EMPTY);
const nextSources: ObservableInputTuple<A> = argsOrArgArray(sources) as any;

return new Observable((subscriber) => {
let sourceIndex = 0;
const subscribeNext = () => {
if (sourceIndex < nextSources.length) {
let nextSource: Observable<A[number]>;
try {
nextSource = innerFrom(nextSources[sourceIndex++]);
} catch (err) {
subscribeNext();
return;
}
const innerSubscriber = new OperatorSubscriber(subscriber, undefined, noop, noop);
nextSource.subscribe(innerSubscriber);
innerSubscriber.add(subscribeNext);
} else {
subscriber.complete();
}
};
subscribeNext();
});
}
@@ -1,15 +1,11 @@
import { Observable } from '../Observable';
import { ObservableInputTuple, OperatorFunction } from '../types';
import { operate } from '../util/lift';
import { innerFrom } from '../observable/innerFrom';
import { argsOrArgArray } from '../util/argsOrArgArray';
import { createOperatorSubscriber } from './OperatorSubscriber';
import { noop } from '../util/noop';
import { onErrorResumeNext as oERNCreate } from '../observable/onErrorResumeNext';

export function onErrorResumeNext<T, A extends readonly unknown[]>(
export function onErrorResumeNextWith<T, A extends readonly unknown[]>(
sources: [...ObservableInputTuple<A>]
): OperatorFunction<T, T | A[number]>;
export function onErrorResumeNext<T, A extends readonly unknown[]>(
export function onErrorResumeNextWith<T, A extends readonly unknown[]>(
...sources: [...ObservableInputTuple<A>]
): OperatorFunction<T, T | A[number]>;

Expand Down Expand Up @@ -85,7 +81,7 @@ export function onErrorResumeNext<T, A extends readonly unknown[]>(
* Observable, but - if it errors - subscribes to the next passed Observable
* and so on, until it completes or runs out of Observables.
*/
export function onErrorResumeNext<T, A extends readonly unknown[]>(
export function onErrorResumeNextWith<T, A extends readonly unknown[]>(
...sources: [[...ObservableInputTuple<A>]] | [...ObservableInputTuple<A>]
): OperatorFunction<T, T | A[number]> {
// For some reason, TS 4.1 RC gets the inference wrong here and infers the
Expand All @@ -94,32 +90,10 @@ export function onErrorResumeNext<T, A extends readonly unknown[]>(
// asserted explicitly.
const nextSources = argsOrArgArray(sources) as unknown as ObservableInputTuple<A>;

return operate((source, subscriber) => {
const remaining = [source, ...nextSources];
const subscribeNext = () => {
if (!subscriber.closed) {
if (remaining.length > 0) {
let nextSource: Observable<A[number]>;
try {
nextSource = innerFrom(remaining.shift()!);
} catch (err) {
subscribeNext();
return;
}

// Here we have to use one of our Subscribers, or it does not wire up
// The `closed` property of upstream Subscribers synchronously, that
// would result in situation were we could not stop a synchronous firehose
// with something like `take(3)`.
const innerSub = createOperatorSubscriber(subscriber, undefined, noop, noop);
nextSource.subscribe(innerSub);
innerSub.add(subscribeNext);
} else {
subscriber.complete();
}
}
};

subscribeNext();
});
return (source) => oERNCreate(source, ...nextSources);
}

/**
* @deprecated Renamed. Use {@link onErrorResumeNextWith} instead. Will be removed in v8.
*/
export const onErrorResumeNext = onErrorResumeNextWith;
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shouldn't onErrorResumeNext stay in this file (onErrorResumeNext.ts) and onErrorResumeNextWith get exported from another file (onErrorResumeNextWith.ts)? This is what's done with combineLatest, concat and others...

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm indifferent about that. That's all internal implementation and has no bearing on users.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we may have issues with the docs builder, though I'm not completely sure. Let me check it.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I can't really make the docs work properly. The code changes from this PR seem to create multiple issues.

One of the issues that this function has is described in #7009.

The other is this:

image

The dgeni didn't properly infer the type. The inferred type seems to be: { <T, A extends any, unknown>(): any; <T, A extends any, unknown>(): any; <T, A extends any, unknown>(): any; }; which doesn't really mean anything.

I will try to dig deeper into the issue tomorrow.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It looks like other constants are having the same issue. E.g. flatMap:

image

So, I don't think this is something we can do about by moving to another file, which would be OK only for consistency reasons, nothing else.

2 changes: 1 addition & 1 deletion src/operators/index.ts
Expand Up @@ -57,7 +57,7 @@ export { mergeWith } from '../internal/operators/mergeWith';
export { min } from '../internal/operators/min';
export { multicast } from '../internal/operators/multicast';
export { observeOn } from '../internal/operators/observeOn';
export { onErrorResumeNext } from '../internal/operators/onErrorResumeNext';
export { onErrorResumeNext } from '../internal/operators/onErrorResumeNextWith';
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change (if you decide to have onErrorResumeNext.ts and onErrorResumeNextWith.ts files):

export { onErrorResumeNext } from '../internal/operators/onErrorResumeNext';
export { onErrorResumeNextWith } from '../internal/operators/onErrorResumeNextWith';

export { pairwise } from '../internal/operators/pairwise';
export { partition } from '../internal/operators/partition';
export { pluck } from '../internal/operators/pluck';
Expand Down