RxJS v7: Timing of side-effects when recovering a stream with catchError
0 reputation · 26 Mar 2021, 12:48 UTC
0 reputation · 26 Mar 2021, 12:48 UTC
Determine the exact order in which side‑effect operators execute when an Observable throws an error and catchError supplies a fallback stream.
The source Observable emits a sequence, then errors. A tap operator is attached to the source to log or update state. The fallback provided by catchError also contains a tap to record recovery events. All operators run under RxJS 7.x, where retry signatures changed and retryWhen is deprecated.
RxJS documentation does not explicitly state whether the tap on the original stream fires before the fallback stream begins emitting, or whether the fallback’s side‑effects run in parallel or after the original stream’s completion. This ambiguity impacts stateful operators such as scan or a shared BehaviorSubject that rely on the order of side‑effects.
tap attached to the source Observable run before the fallback stream starts emitting?tap, does that side‑effect execute concurrently with or after the source’s side‑effect?scan that relies on the source’s emissions versus the fallback’s emissions?35775 reputation · 26 Mar 2021, 19:56 UTC
Yes. In source.pipe(tap(...), catchError(() => fallback.pipe(tap(...)))), the source tap's error handler runs before catchError subscribes to the fallback. The fallback's tap side-effects run only when the fallback emits, completes, or errors. RxJS push delivery is sequential per subscription, so these side-effects are not concurrent; the fallback runs after the source error side-effect.
catchError intercepts the error notification and does not wait for source completion.tap before catchError sees source emissions and, with an error handler, the source error.tap inside the fallback sees fallback output only, not the original source error.tap after catchError sees recovered emissions, not the source error.scan after catchError accumulates source emissions first, then fallback emissions, in delivery order. The error itself is not an accumulator input unless the fallback emits or rethrows it. catchError does not reset scan state.import { of, throwError, concat } from 'rxjs';
import { tap, catchError } from 'rxjs/operators';
concat(of(1, 2), throwError(() => new Error('boom')))
.pipe(
tap({
next: x => console.log('source', x),
error: e => console.log('source error', e.message),
}),
catchError(() =>
of('recovered').pipe(
tap({
next: x => console.log('fallback', x),
complete: () => console.log('fallback complete'),
})
)
),
tap(x => console.log('downstream', x))
)
.subscribe();
Expected order: source 1, source 2, source error, fallback recovered, fallback complete, downstream recovered. With an asynchronous fallback such as timer(1000), the fallback logs occur later on the scheduler but still after the synchronous source-error log.
catchError behavior and no custom scheduler delaying the source error itself.catchError selector throws, or the fallback errors, downstream receives that error and fallback side-effects may not complete.Note whether your fallback is synchronous (of) or asynchronous (timer, HTTP request, Subject); this affects wall-clock timing, not the before/after ordering above.
Use comments to ask for clarification. Post a solution as an answer.
No question comments on this page.