RxJS: как выдать ошибку, а затем поймать ее?

Рассмотрим следующую наблюдаемую ниже, которая выдает значения от 1 до 4. Допустим, ошибка выдается, когда значение равно 3. Эта ошибка может быть обнаружена в операторе catchError или внутри subscribe. Я считаю, что это зависит от конкретного варианта использования, позволяете ли вы всплывать ошибке полностью до подписчика или ее следует обрабатывать где-то выше по течению от подписчика.

of(1, 2, 3, 4).pipe(
  // Throw error when value is 3
  tap(value => { if(value === 3) throw new Error('Oops!') }),
  catchError(err => {
    console.log('Catch error in operator', err);

    // You can rethrow the same error or a new error
    // return throwError(err);

    // Or you can handle the error and return a new observable
    return of(3)
  })
).subscribe(
  value => console.log(value),
  // If `catchError` returns a new observable, then the error 
  // function here will not be called
  err => console.log('Catch error within subscribe', err),
  () => console.log('Done!')
)

Обратите внимание, что в этом примере, даже если ошибка обрабатывается, наблюдаемая завершается, и значение 4 никогда не выдается. Если вы хотите, чтобы наблюдаемое оставалось активным при возникновении ошибки, посмотрите этот ответ StackOverflow.

вместо:

public getData() {
  return this._getBoolean().pipe(
    throwError(bool => bool ? new Error('Wrong server answer, stream stopped') : null),
    catchError(err => console.warn(err)),
    switchMap(bool => this._getData())
  );
}

почему бы не что-то вроде:

public getData() {
  return this._getBoolean().pipe(
    tap(result => !result && throwError('Wrong server answer, stream stopped')),
    switchMap(bool => this._getData()),
    catchError(err => console.warn(err))
  );
}

Я не уверен, правильно ли я понимаю вашу проблему, но вы можете заменить

    console.warn('Wrong server answer, stream stopped');
    return false;

С

   Observable.throw('Some error cause')

А затем перехватите его с помощью ближайшего блока catch в вашем потоке, что дает вам в основном изменение: - Остановить поток, если вы повторно выдаете ошибку - Перезапустите его, если вы вернете наблюдаемый ввод - верните полностью новый наблюдаемый

public getData() {
  return this._getBoolean().pipe(
    filter(bool => {
      if(!bool) {
        console.warn('Wrong server answer, stream stopped');
        //return false;
        Observable.throw('I got false where I expected true')
      }
      return true;
    }),
    switchMap(bool => this._getData())
  );
}

А потом:

getData()
.any()
.operator()
.you()
.wish()
.catch(e => {
  /* Here stream will be terminated on thrown error */
})