Keep moving forward with RXJS by subscribing to concat after encountering an error

Question

Keep moving forward with RXJS by subscribing to concat after encountering an error

I am facing a challenge with an array of observables that need to fire off sequentially. My goal is for the observer to catch any errors, log them, and continue observing without restarting or completing.

Currently, when an error occurs, the observer stops instead of continuing as desired.

import * as Rx from "rxjs";
const source = [
  Rx.Observable.from("1").delay(200),
  Rx.Observable.from("2").delay(150),
  Rx.Observable.throw("error"),
  Rx.Observable.from("3").delay(124),
  Rx.Observable.from("4").delay(201),
];
let sSource = Rx.Observable.concat(...source);
sSource.subscribe((v) => {console.log(v)}, (e) => {console.log(e)});

Current output:

1
2
error

Expected output:

1
2
error
3
4

We have tried adding individual catch handlers to each observable in the source array but feel there should be a more elegant solution to this problem. If necessary, I can share our current solution.

typescript rxjs rxjs5 rxjs6

Answer 1

Answer №1

To handle errors in multiple source observables, you can utilize the catch operator and implement error logging within it. Here's an example:

const sources = [
  Rx.Observable.from("1").delay(200),
  Rx.Observable.from("2").delay(150),
  Rx.Observable.throw("error"),
  Rx.Observable.from("3").delay(124),
  Rx.Observable.from("4").delay(201),
];
const sourcesWithCatch = sources.map(s => s.catch(e => {
  console.log(e);
  return Rx.Observable.empty();
}));
const concatted = Rx.Observable.concat(...sourcesWithCatch);
concatted.subscribe(v => console.log(v));

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"></script>

Answer 2

To handle errors in multiple source observables, you can utilize the catch operator and implement error logging within it. Here's an example:

const sources = [
  Rx.Observable.from("1").delay(200),
  Rx.Observable.from("2").delay(150),
  Rx.Observable.throw("error"),
  Rx.Observable.from("3").delay(124),
  Rx.Observable.from("4").delay(201),
];
const sourcesWithCatch = sources.map(s => s.catch(e => {
  console.log(e);
  return Rx.Observable.empty();
}));
const concatted = Rx.Observable.concat(...sourcesWithCatch);
concatted.subscribe(v => console.log(v));

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"></script>

Answer 3

Answer №2

Keep in mind that Rxjs 5 introduces the onErrorResumeNext function which mimics the behavior of the Visual Basic On Error Resume Next statement.

You can find this example in the official documentation.

var source = Rx.Observable.onErrorResumeNext(
  Rx.Observable.just(42),
  Rx.Observable.throw(new Error()),
  Rx.Observable.just(56),
  Rx.Observable.throw(new Error()),
  Rx.Observable.just(78)
);

var subscription = source.subscribe(
  data => console.log(data)
);
// => 42
// => 56
// => 78

Answer 4

Keep in mind that Rxjs 5 introduces the onErrorResumeNext function which mimics the behavior of the Visual Basic On Error Resume Next statement.

You can find this example in the official documentation.

var source = Rx.Observable.onErrorResumeNext(
  Rx.Observable.just(42),
  Rx.Observable.throw(new Error()),
  Rx.Observable.just(56),
  Rx.Observable.throw(new Error()),
  Rx.Observable.just(78)
);

var subscription = source.subscribe(
  data => console.log(data)
);
// => 42
// => 56
// => 78

Keep moving forward with RXJS by subscribing to concat after encountering an error

Answer №1

Answer №2

Similar questions

NGRX Store: Unable to modify the immutable property '18' of the object '[object Array]'

`How can I sort information based on a chosen parameter?`

Encountered an issue with the Dynamic Form: TypeError - The property 'value' is undefined and cannot be read

The type 'string' cannot be assigned to type 'ImageSourcePropType'

Node.js is throwing the error "error TS18003: Config file does not contain any inputs."

The issue arises when trying to convert t to a float in TensorFlow within an Angular environment because

"Transforming a callback function to an asynchronous function results in an error

Tips for transforming the inner elements of each Object emitted by an observable prior to releasing the overarching observable

Uncomplicating RxJs Operators: Decoding switchMap and combineLatest

What is the best way to obtain the dimensions of an image in Angular 2 (or newer) before uploading it to the server, and can this be accomplished without utilizing jQuery?

Changing the value of an object in Angular can be achieved by utilizing the two

Ways to incorporate type into an object comprised of n element in Typescript

Utilizing Service within Express Router

Angular - Dividing Values within Input Arrays

Error in Angular 2 component when loading background images using relative URLs from an external CSS skin

Issue encountered while executing npm install due to rxjs version discrepancy

Creating TypeScript declarations for the `enhanced-resolve` library: A step-by-step guide

Typescript error encountered while attempting to fetch repository from Redis OM

Which rxjs operator emits a single value when any of the source observables emits a value?

Converting constants into JavaScript code