Rxjs: Feature request: add `onErrorResumeNext`

[danhyun]

According to https://github.com/ReactiveX/rxjs/blob/master/MIGRATION.md we're to file an issue or PR requesting feature additions to RxJS5.

I'd like to have something like onErrorResumeNext http://xgrommx.github.io/rx-book/content/observable/observable_methods/onerrorresumenext.html or the ability to continue with the next emitted item.

In my specific usecase I'm doing something like this:

Rx.Observable.from([1, 2, 3, 4])
  .map(i => {
    if (i == 2) {
       throw new Error('no 2s');
    }
    return i;
  })
  //.onErrorResumeNext(somefunc) // short circuit to subscriber's on error
  .do(i => console.log('map 1: ' + i))
  .map(i => i + 1)
  //.onErrorResumeNext(somefunc) // short circuit to subscriber's on error
  .do(i => console.log('map 2: ' + i))
  .map(i => {
    if (i & 0 == 0) {
      throw new Error('no evens');
    }
    return i * 2;
  })
  .do(i => console.log('map 3: ' + i))
//.onErrorResumeNext(somefunc) // short circuit to subscriber's on error
  .subscribe(
    i => console.log('found ' + i),
    e => console.log('error ' + e),
    () => console.log('done')
  );

Related JSBin http://jsbin.com/monuyegatu/1/edit?html,js,console

[trxcllnt]

Have you seen catch (and tests)?

[danhyun]

Thanks @trxcllnt those are awesome resources.

I'm still having a bit of trouble here, so in my use case I'm creating a cold observable from an array. Trying to use catch it looks like it replaces the rest of the sequence rather than replacing just that single item. Given that, is there any way to get the remaining sequence from catch?

  .catch((e, o) => {
    console.log('Handling error: ' + e);
    return Rx.Observable.from([1,2,3,4]); // any way to recover and resume the existing cold observable sequence?
  })

Since my Observable has a number of transformations I'd like to shortciruit the processing of this current item and just proceed to the next one. I haven't been able to tease out how I would do this right now without keeping track of the items I've processed or without something like Observable#onErrorResumeNext()

[benlesh]

Yeah, we need to add onErrorResumeNext

[danhyun]

Thanks!

[omerts]

Hey,

I can't understand how onErrorResumeNext solves @danhyun's issue.
The source observable Rx.Observable.from([1, 2, 3, 4]), gets terminated after the error, while he (and we too), want it to continue with the next emission.
Maybe I am doing something wrong, could someone explain please?
Here is a fiddle: https://jsfiddle.net/omerts/c9ky94ac/

[omerts]

Just to add, if I understood the RXJava operator right, what we need is: onExceptionResumeNext
onExceptionResumeNext - instructs an Observable to continue emitting items after it encounters an exception (but not another variety of throwable)

[gunar]

You can use .catch((e, $) => $) to continue with the next value.
I've heard this is not safe to use with all Observables as it may trigger infinite loops, but seems safe for Subjects.

'use strict'

const Rx = require('rxjs')

const $ = new Rx.Subject()

const throwOnEven = x => {
  if (x % 2 === 0) throw new Error(x)
  return x
}

$
  .map(throwOnEven)
  .catch((e, stream) => stream)
  .subscribe(x => {
    console.log(`odd: ${x}`)
    // this is uncatched, still
    if (x > 4) throw Error('shit')
  }, e => {
    console.log('fatal error', e)
  })

try {
  $.next(1)
  $.next(2)
  $.next(3)
  $.next(4)
  $.next(5)
} catch (e) {
  console.log({outer:e}) // { outer: Error('shit') }
}
// odd: 1
// odd: 3
// odd: 5
// { outer: Error('shit') }

[lock[bot]]

This thread has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.