I'm struggling to handle errors with Rx in the manner I expected.

When I have an Observable (for example, from a click stream) and an exception occurs I want to catch it but carry on. All the solutions I have tried catch the error then end the Observable. Is there a way to catch and carry on?

As an example the Observable below will emit "1", then "Error", but never "3".

var range = Rx.Observable.range(1,3)
    .map(function(i){
      if(i === 2){
        throw "Error";
      } else {
        return i;
      }
    })
    .catch(function(e){
      return Rx.Observable.return(e)
    });

range.subscribe(function(i){
  console.log(i)
});

I'm struggling to handle errors with Rx in the manner I expected.

When I have an Observable (for example, from a click stream) and an exception occurs I want to catch it but carry on. All the solutions I have tried catch the error then end the Observable. Is there a way to catch and carry on?

As an example the Observable below will emit "1", then "Error", but never "3".

var range = Rx.Observable.range(1,3)
    .map(function(i){
      if(i === 2){
        throw "Error";
      } else {
        return i;
      }
    })
    .catch(function(e){
      return Rx.Observable.return(e)
    });

range.subscribe(function(i){
  console.log(i)
});

• javascript
• system.reactive
• rxjs

• The only way to carry on with the source sequence is not to throw. Could you tell us what your real use case is? – Oliver Weichhold Commented May 22, 2015 at 17:43
• The observable contract is that it must emit to an observer the pattern: (OnNext)* (OnCompleted | OnError). So the moment a sequence emits OnError it ends. – Timothy Shields Commented May 23, 2015 at 17:44
• @OliverWeichhold the real case is a click event observable, which flatMap() to a http request observable. If/when the http request fails, the click observable is no longer emitting. – MaxWillmott Commented May 26, 2015 at 9:56

1 Answer 1

While your expected behavior cannot be achieved because of the Observable contract (OnNext)* (OnCompleted|OnError), there are practical ways of properly working around this by introducing a hot Observable.

let hotRange = Rx.Observable.range(1,3).publish();

let safeRange = hotRange
  .map(function (i) {
    if (i === 2) {
      throw "Error";
    } else {
      return i;
    }
  })
  .retry();

safeRange.subscribe(i => console.log(i));
hotRange.connect();

See the JSBin. The range Observable you mentioned in the question is a cold Observable. It behaves as a movie, so if an error happens and we resubscribe, we need to subscribe from the beginning of the "movie", that is, 1 then "Error".

You probably had an implicit assumption that Rx.Observable.range(1, 3) was a live Observable, i.e., "hot". Since it isn't, I made hotRange above using publish(). This way, it will emit its events independently of its subscribers. If we want to be able to "carry on" after an error, we need our source ("hotRange") to be without errors. That's why range.map( ) isn't the hot Observable. retry() will catch errors on hotRange.map( ) and replace it with hotRange.map( ). Because hotRange is hot, every execution of retry() will be different, since it doesn't remember previous values emitted by hotRange. So when the error caused by 2 is replaced by hotRange.map( ) in the retry, hotRange will subsequently emit 3, and pass the map function without errors.