118 lines
4.7 KiB
JavaScript
118 lines
4.7 KiB
JavaScript
import { FromObservable } from '../observable/FromObservable';
|
|
import { isArray } from '../util/isArray';
|
|
import { OuterSubscriber } from '../OuterSubscriber';
|
|
import { subscribeToResult } from '../util/subscribeToResult';
|
|
/* tslint:enable:max-line-length */
|
|
/**
|
|
* When any of the provided Observable emits an complete or error notification, it immediately subscribes to the next one
|
|
* that was passed.
|
|
*
|
|
* <span class="informal">Execute series of Observables no matter what, even if it means swallowing errors.</span>
|
|
*
|
|
* <img src="./img/onErrorResumeNext.png" width="100%">
|
|
*
|
|
* `onErrorResumeNext` is an operator that accepts a series of Observables, provided either directly as
|
|
* arguments or as an array. If no single Observable is provided, returned Observable will simply behave the same
|
|
* as the source.
|
|
*
|
|
* `onErrorResumeNext` returns an Observable that starts by subscribing and re-emitting values from the source Observable.
|
|
* When its stream of values ends - no matter if Observable completed or emitted an error - `onErrorResumeNext`
|
|
* will subscribe to the first Observable that was passed as an argument to the method. It will start re-emitting
|
|
* its values as well and - again - when that stream ends, `onErrorResumeNext` will proceed to subscribing yet another
|
|
* Observable in provided series, no matter if previous Observable completed or ended with an error. This will
|
|
* be happening until there is no more Observables left in the series, at which point returned Observable will
|
|
* complete - even if the last subscribed stream ended with an error.
|
|
*
|
|
* `onErrorResumeNext` can be therefore thought of as version of {@link concat} operator, which is more permissive
|
|
* when it comes to the errors emitted by its input Observables. While `concat` subscribes to the next Observable
|
|
* in series only if previous one successfully completed, `onErrorResumeNext` subscribes even if it ended with
|
|
* an error.
|
|
*
|
|
* Note that you do not get any access to errors emitted by the Observables. In particular do not
|
|
* expect these errors to appear in error callback passed to {@link subscribe}. If you want to take
|
|
* specific actions based on what error was emitted by an Observable, you should try out {@link catch} instead.
|
|
*
|
|
*
|
|
* @example <caption>Subscribe to the next Observable after map fails</caption>
|
|
* Rx.Observable.of(1, 2, 3, 0)
|
|
* .map(x => {
|
|
* if (x === 0) { throw Error(); }
|
|
return 10 / x;
|
|
* })
|
|
* .onErrorResumeNext(Rx.Observable.of(1, 2, 3))
|
|
* .subscribe(
|
|
* val => console.log(val),
|
|
* err => console.log(err), // Will never be called.
|
|
* () => console.log('that\'s it!')
|
|
* );
|
|
*
|
|
* // Logs:
|
|
* // 10
|
|
* // 5
|
|
* // 3.3333333333333335
|
|
* // 1
|
|
* // 2
|
|
* // 3
|
|
* // "that's it!"
|
|
*
|
|
* @see {@link concat}
|
|
* @see {@link catch}
|
|
*
|
|
* @param {...ObservableInput} observables Observables passed either directly or as an array.
|
|
* @return {Observable} An Observable that emits values from source Observable, but - if it errors - subscribes
|
|
* to the next passed Observable and so on, until it completes or runs out of Observables.
|
|
* @method onErrorResumeNext
|
|
* @owner Observable
|
|
*/
|
|
export function onErrorResumeNext(...nextSources) {
|
|
if (nextSources.length === 1 && isArray(nextSources[0])) {
|
|
nextSources = nextSources[0];
|
|
}
|
|
return (source) => source.lift(new OnErrorResumeNextOperator(nextSources));
|
|
}
|
|
/* tslint:enable:max-line-length */
|
|
export function onErrorResumeNextStatic(...nextSources) {
|
|
let source = null;
|
|
if (nextSources.length === 1 && isArray(nextSources[0])) {
|
|
nextSources = nextSources[0];
|
|
}
|
|
source = nextSources.shift();
|
|
return new FromObservable(source, null).lift(new OnErrorResumeNextOperator(nextSources));
|
|
}
|
|
class OnErrorResumeNextOperator {
|
|
constructor(nextSources) {
|
|
this.nextSources = nextSources;
|
|
}
|
|
call(subscriber, source) {
|
|
return source.subscribe(new OnErrorResumeNextSubscriber(subscriber, this.nextSources));
|
|
}
|
|
}
|
|
class OnErrorResumeNextSubscriber extends OuterSubscriber {
|
|
constructor(destination, nextSources) {
|
|
super(destination);
|
|
this.destination = destination;
|
|
this.nextSources = nextSources;
|
|
}
|
|
notifyError(error, innerSub) {
|
|
this.subscribeToNextSource();
|
|
}
|
|
notifyComplete(innerSub) {
|
|
this.subscribeToNextSource();
|
|
}
|
|
_error(err) {
|
|
this.subscribeToNextSource();
|
|
}
|
|
_complete() {
|
|
this.subscribeToNextSource();
|
|
}
|
|
subscribeToNextSource() {
|
|
const next = this.nextSources.shift();
|
|
if (next) {
|
|
this.add(subscribeToResult(this, next));
|
|
}
|
|
else {
|
|
this.destination.complete();
|
|
}
|
|
}
|
|
}
|
|
//# sourceMappingURL=onErrorResumeNext.js.map
|