rxjs how to handle error in subscriber's next

Question

I'm using rxjs in a library to expose an Observable that clients can subscribe to to consume messages. I want to be able to react appropriately if the subscriber's next function throws an error. However, I'm not seeing any obvious way to detect that. For example:

const observable = new Observable<string>((subscriber) => {
  subscriber.next('first')
  subscriber.next('second')
  subscriber.complete()
})

observable.subscribe(() => {
  throw new Error('oh no!')
})

I have tried all of the following, but the errors are bubbled all the way up to a global scope that's surfaced either in an onUnhandledError function provided to the global config, or in absence of that, the node process's unhandledException event.

process.on('uncaughtException', (error) => {
  console.error('IN UNCAUGHT EXCEPTION HANDLER', error.message)
})

export function main() {
  try {
    const observable = new Observable<string>((subscriber) => {
      try {
        subscriber.next('first')
      } catch (error) {
        console.error('IN OBSERVABLE CATCH', error.message)
      }
      subscriber.complete()
    }).pipe(
      catchError((error) => {
        console.error('CATCHERROR PIPE', error.message)
        return of('there was an error!!!!')
      }),
    )

    observable.subscribe({
      next: (_value) => {
        throw new Error('oh no!')
      },
      error: (error) => {
        console.error('IN OBSERVER ERROR HANDLER', error.message)
      },
      complete: () => console.log('complete!'),
    })
  } catch (error) {
    console.error('IN MAIN CATCH', error.message)
  }
}

This logs:

complete!
IN UNCAUGHT EXCEPTION HANDLER oh no!

The docs don't make a big fuss about ensuring that subscribers avoid throwing errors at all costs, but I don't see a standard mechanism for handling it short of some sort of "observer wrapper" (that gets a bit ugly with the overloads).

Answer 1

Turns out that there is effectively no way to handle errors from observers.

When using Observable.subscribe, it wraps your observer functions in a "SafeSubscriber". This then wraps the supplied functions with a ConsumerObserver. This wraps each in a try/catch that, upon errors, either:

• sends them to an optional onUnhandledError function you can supply to the rxjs config
• sends them "into the ether" to be picked up by the node process

There is no context, and no way to hook into it. Effectively, errors from next, error, or complete handlers just silently disappear.