Handling errors in RxJS, RxSwift or RxJava ~ Yair Carreno

Handling errors in RxJS, RxSwift or RxJava

The key to effective error management is to identify the strategy to react to these scenarios adequately.

When an error occurs in the stream, it could react in the following ways:

Scenario 1:

Do nothing, do not handle the error, and allow the exception to propagate to the Observer with immediate termination of the operator chain execution and shutdown of the stream. That is the one that should never be followed.

Scenario 2:

Capture the error, control it, and assign a custom value without turning off the stream, despite the immediate completion of the operator chain’s execution.

Scenario 3:

Catch the error, log it, and ignore it to suspend the operator chain’s execution without turning off the stream.

Scenario 4:

An error occurs, the task execution is retried a limited number of times (n times). If the error persists after retries, the operator chain’s execution is captured, controlled, and suspended without turning off the stream.

Scenario 5:

An error occurs, the task execution is retried a limited number of times (n times) every specific time window (period). If the error persists after retries, the operator chain’s execution is captured, controlled, and suspended without turning off the stream.

Rx provides operators that allow designing different strategies to apply the mentioned scenarios. Below are examples of the recommended strategies. It is emphasized that scenario number 1 should be avoided.

How to do it with RxJS?

Example code for scenario 1 with RxJS:

network.getToken("api-key")

    .pipe(

        concatMap(token => this.cache.storeTokenWithError(token)),

        concatMap(saved => this.network.getUser(username))

)

    .subscribe(user => console.log(user));

In this case, storeTokenWithError task will generate an error. The result is emission suspension, and only the onError contract is executed, not even onComplete. Remember that onError and onComplete are mutually exclusive, or one runs, or the other runs but not both.

Example code for scenario 2 with RxJS:

network.getToken("api-key")
    .pipe(
        concatMap(token => this.cache.storeTokenWithError(token)),
        concatMap(saved => this.network.getUser(username)),
        catchError(error => of(new User())))
    .subscribe(
        user => console.log('next:', user),
        error => console.log('error:', error),
        () => console.log('completed'));

In this case, the error is handled and customized (User is propagated empty or with default values).

Also, the contracts onNext and onComplete are reached. It should also be noted that the getUser task cannot be executed since once the error is generated in storeTokenWithError, the execution of the operator chain is suspended.

Does the location of the error operator matter?

Yes. Capturing the error is done at the point where the operator is defined. In a chain of operators, it is recommended to locate it just before the subscription.

Example code for scenario 3 with RxJS:

network.getToken("api-key")
    .pipe(
        concatMap(token => this.cache.storeTokenWithError(token)),
        concatMap(saved => this.network.getUser(username)),
        catchError(error =>
            empty()
                .pipe(tap({ complete: () => console.log(error) }))
        ))
    .subscribe(
        user => console.log('next:', user),
        err => console.log('error:', err),
        () => console.log('completed'));

In this case, when the error occurs, the trace is left in logs, and the emission continues, ignoring the value.

Example code for scenario 4 with RxJS:

network.getToken("api-key")
    .pipe(
        concatMap(token => this.cache.storeTokenWithError(token)),
        concatMap(saved => this.network.getUser(username)),
        retry(2),
        catchError(error =>
            empty()
                .pipe(tap({ complete: () => console.log(error) }))
        ))
    .subscribe(
        user => console.log('next:', user),
        err => console.log('error:', err),
        () => console.log('completed'));

Similar to the previous scenario, only the retry operator is added. It is crucial always to specify the number of attempts since if this parameter is not specified, the retry could be permanent and counterproductive for the application’s performance.

Example code for scenario 5 with RxJS:

network.getToken("api-key")
    .pipe(
        concatMap(token => this.cache.storeTokenWithError(token)),
        concatMap(saved => this.network.getUser(username)),
        retryWhen(errors => errors.pipe(
            tap(() => console.log('retrying...')),
            delay(2000),
            take(4),
            concat(throwError('SampleError!'))
        )),
        catchError(error =>
            empty()
                .pipe(tap({ complete: () => console.log(error) }))
        ))
    .subscribe(
        user => console.log('next:', user),
        err => console.log('error:', err),
        () => console.log('completed'));

It is an advanced technique where you will retry executing the task a limited number of times. If no retry is successful, then the error is handled, and the emission ends.

And what about RxSwift?

Well, the same scenarios can apply with RxSwift. Below is the implementation of each scenario using RxSwift:

Example code for scenario 1 with RxSwift:

network.getToken("api-key")

.concatMap { token in self.cache.storeTokenWithError(token) }

.concatMap { saved in self.network.getUser(username) }

.subscribe(onNext: { user in print("next:", user) },

onError: { error in print("error:", error) },

onCompleted: { print("completed") } )

.disposed(by: disposeBag)

Example code for scenario 2 with RxSwift:

network.getToken("api-key")

.concatMap { token in self.cache.storeTokenWithError(token) }

.concatMap { saved in self.network.getUser(username) }

.catchError({ error in Observable.just(User()) })

.subscribe(onNext: { user in print("next:", user) },

onError: { error in print("error:", error) },

onCompleted: { print("completed") } )

.disposed(by: disposeBag)

Example code for scenario 3 with RxSwift:

network.getToken("api-key")

.concatMap { token in self.cache.storeTokenWithError(token) }

.concatMap { saved in self.network.getUser(username) }

.catchError({ error in Observable.empty()

.do(onCompleted: { print(error) }) })

.subscribe(onNext: { user in print("next:", user) },

onError: { error in print("error:", error) },

onCompleted: { print("completed") } )

.disposed(by: disposeBag)

Example code for scenario 4 with RxSwift:

network.getToken("api-key")

.concatMap { token in self.cache.storeTokenWithError(token) }

.concatMap { saved in self.network.getUser(username) }

.retry(2)

.catchError({ error in Observable.empty()

.do(onCompleted: { print(error) }) })

.subscribe(onNext: { user in print("next:", user) },

onError: { error in print("error:", error) },

onCompleted: { print("completed") } )

.disposed(by: disposeBag)

Example code for scenario 5 with RxSwift:

network.getToken("api-key")

.concatMap { token in self.cache.storeTokenWithError(token) }

.concatMap { saved in self.network.getUser(username) }

.retryWhen{ errors in errors

.do(onNext: { ignored in print("retrying...") })

.delay(.seconds(2), scheduler: MainScheduler.instance)

.take(4)

.concat(Observable.error(SampleError()))}

.catchError({ error in Observable.empty()

.do(onCompleted: { print(error) }) })

.subscribe(onNext: { user in print("next:", user) },

onError: { error in print("error:", error) },

onCompleted: { print("completed") } )

.disposed(by: disposeBag)

And what about RxJava?

Example code for scenario 1 with RxJava:

network.getToken("api-key")

.concatMap(token -> cache.storeTokenWithError(token))

.concatMap(saved -> network.getUser(username))

.subscribe(element -> Log.d(TAG, "" + element),

throwable -> Log.e(TAG, "error:" + throwable),

() -> Log.d(TAG, "completed"))

Example code for scenario 2 with RxJava:

network.getToken("api-key")

.concatMap(token -> cache.storeTokenWithError(token))

.concatMap(saved -> network.getUser(username))

.onErrorResumeNext(throwable -> Observable.just(new User()))

.subscribe(user -> Log.d(TAG, "next: " + user),

throwable -> Log.e(TAG, "error: " + throwable),

() -> Log.d(TAG, "completed"))

Example code for scenario 3 with RxJava:

network.getToken("api-key")

.concatMap(token -> cache.storeTokenWithError(token))

.concatMap(saved -> network.getUser(username))

.onErrorResumeNext(throwable -> {

Log.d(TAG, Objects.requireNonNull(throwable.getMessage()));

return Observable.empty();

})

.subscribe(user -> Log.d(TAG, "next: " + user),

throwable -> Log.e(TAG, "error: " + throwable),

() -> Log.d(TAG, "completed"))

Example code for scenario 4 with RxJava:

network.getToken("api-key")

.concatMap(token -> cache.storeTokenWithError(token))

.concatMap(saved -> network.getUser(username))

.retry(2)

.onErrorResumeNext(throwable -> {

Log.d(TAG, Objects.requireNonNull(throwable.getMessage()));

return Observable.empty();

})

.subscribe(user -> Log.d(TAG, "next: " + user),

throwable -> Log.e(TAG, "error: " + throwable),

() -> Log.d(TAG, "completed"))

Example code for scenario 5 with RxJava:

network.getToken("api-key")

.concatMap(token -> cache.storeTokenWithError(token))

.concatMap(saved -> network.getUser(username))

.retryWhen(errors -> errors

.doOnNext(ignored -> Log.d(TAG, "retrying..."))

.delay(2, TimeUnit.SECONDS)

.take(4)

.concatWith(Observable.error(new Throwable())))

.onErrorResumeNext(throwable -> {

Log.d(TAG, Objects.requireNonNull(throwable.getMessage()));

return Observable.empty();

})

.subscribe(user -> Log.d(TAG, "next: " + user),

throwable -> Log.e(TAG, "error: " + throwable),

() -> Log.d(TAG, "completed"))

If you want to find more information about RxJS, RxSwift or RxJava, I recommend you check the guide book:

The Clean Way to Use Rx. Also the Spanish version.

Happy reading.

Yair Carreno