Writing a Custom RxJS Operator

Cover Image

While working with rxjs, there are plenty of operators that one can use within the .pipe operator of an observable. Just take a look at the api reference here and you'll realize that rxjs provides all the operators that you need for most cases.

However, what if you needed to write your custom operator to transform data the way you wanted, or tweak the captureError operator to handle the error in a certain way and return something else in case an error happened.

In an application I was working on, for which I was using Sentry to handle any errors, I wanted to have the captureError to send the error message to Sentry, and return fallback data in case an error happened.

import { captureException } from '@sentry/angular'
import { BehaviorSubject, Observable, OperatorFunction, catchError, tap } from 'rxjs'
 
export function consume<T>(consumer: BehaviorSubject<T>, fallback$: Observable<T>): OperatorFunction<T, T> {
    return (source$: Observable<T>) => {
        return source$.pipe(
            catchError((error) => {
                captureException(error)
                return fallback$
            }),
            tap(consumer),
        )
    }
}

The operator above is written in the form of a curried function, that accepts two initial inputs: consumer - which in this case was a behavior subject that stores the current value in the observable stream, and fallback$ which is the data to return in case an error happens. If no error happens, the captureError rxjs operator isn't called. In all cases, either the data from the observable stream or the fallback data is passed on to the consuming behavior Subject.

And there we have it! Our own custom RxJS operator 🧞‍♂️

Links