How can I catch an RxJS error only if it is unhandled

Question

I am building an application using RxJS and one of the interesting problems I have come across is how I can implement an operator that will catch errors only when they are unhandled by the rest of the pipeline ahead of it. I'll call my operator catchUnhandledError for now.

The catchError operator is loosely similar (non stream) to

try {
    // stream  
} catch (err) {
    // catchError handler
    // new stream
}

What I am trying to implement resembles the following

try {
    // stream
} catch (err) {
    try {
        // original stream
    } catch (err2) {
        // catchUnhandledError handler
        // new stream
    }
}

The key takeaway here is that the new operator only catches errors that other operators do not catch further down the pipeline.

I appreciate that piped operators essentially wrap observables similar to how middleware works in popular application pipelines which means "pushing the pipe to the end" is nonsensical.

My stream (simplified) is created as follows.

combineLatest([ page, filter ]).pipe(
    switchMap(([ page, { search, order }]) =>
        apiQuery(search, order, page).pipe(
            map(/* combine response with filters */),
            catchError(/* ONLY handle http status 422 Unprocessable Entity for validation */),
            catchError(/* ONLY handle http status 409 Conflict for version conflicts *)
        )
    )
);

Where api.query returns:

function errorHandler(errorMessage?: string) {
    return <T>(source: Observable<T>): Observable<T> => 
        source.pipe(
            /* replace with new catchUnhandledError operator */
            catchError(() => {
                /* More complex code than this to log the error */
                alert('unhandled error');

                return EMPTY;
            })  
        );
}

function apiQuery(search: string, order: string, page: number) {
    /* uses parameters to generate ajax payload */
    return ajax({
        url: 'url', 
        method: 'GET'
    }).pipe(
        map(r => r.response as T),
        errorHandler('An error message')
    );
}

The problem is the validation / request specific error handlers never get called because the generic errorHandler takes precedence. A couple of work arounds I will have to use if this is not possible:

• Pass everything in the pipe as a parameter / callback (convoluted)
• Map success and errors to a single object then check .success (convoluted)
• Copy and paste the generic errorHandler to every single place I call my api (duplicated)

Does anyone have any ideas that will prevent me from having to have convoluted or duplicated code?

Answer 1

For the unhandled observable error handling you can use:

import { config } from 'rxjs'

config.onUnhandledError = (err: any) => console.error(err)

Answer 2

The first thing that comes to mind is to create a custom operator that will be given a callback function with which you can determine whether the catchError should handle this or should pass it along further in the stream.

const catchUnhandledError = (shouldHandle: (err: any) => boolean) => 
  (source: Observable<any>) => defer(
    () => source.pipe(
      catchError(err => shouldHandle(err) ? of('handled') : throwError(err) /* pass along the error */)
    )
  )