RxJS: Observable.never () сохраняет подписку

Question

Я использую rxjs 5.5.6.

Я создал этот код, чтобы показать поведение:

Observable.of(1, 2)
    .do(a => {
        console.log(a);
        let d:string = null;
        let r = d.length;  // it raises an null exception
    })
    .catch(() => {
        console.log("error catched");  //exception is capture here
        return Observable.never();
    })
    .subscribe();

Я ожидал, что результат будет:

1
error catched
2
error catched

Однако вывод:

1
error catched

Это означает, что подписка прекращается, несмотря на то, что Observable.never() возвращается в цепочке .catch(...) метода.

Есть идеи?

РЕАЛЬНЫЙ СЛУЧАЙ

this.subs = Observable
    .merge(this.$searchQuery, this.$lazyQuery)
    .do(() => this.loadingPage())
    .map(filter => this.buildURL(user, app, filter))
    .switchMap(url => this.service.getItemsFromService(url))
    .map(response => this.buildPage(response))
    .do(page => this.loadedPage(page))
    .catch(() => {
        this.loadedPage(pojo.Page.EMPTY);
        return Observable.never();
    })
    .takeUntil(this.$unsubscribe)
    .subscribe();

Answer 1

Да, все объяснения @frido верны.С этим ответом я хотел бы добавить:

Если вы хотите зафиксировать любую ошибку, возникающую в какой-либо конкретной наблюдаемой (например, HTTP-запрос), то вам нужно обработать ее в этой конкретной наблюдаемой ошибкой.

  let correct = Observable.of("correct")
  let inCorrect = Observable.throw('inCorect')

  let obs = [inCorrect, correct];
  let handledObs = obs.map(eachObs => {
    return eachObs.catch((e) => {
      console.log("Individual handler, Good Catch!");
      return Observable.of("I am tampered");
    })
  })

  forkJoin(...handledObs)
  .do(a => {
      console.log(a);
  })
  .catch(() => {
      console.log("error catched");
      return Observable.never();
  })
  .subscribe(data => {
    console.log(`data`, data)
  },(e) => {
      console.log(`error`, e)
  });

}

См. Пример здесь: https://stackblitz.com/edit/angular-7mmhn7?file=src/app/app.component.ts

---

РЕДАКТИРОВАТЬ

Однако, когда я смотрю на ваш код, мне кажется, что выпопытка что-то записать в журнал и возвращенные данные могут не иметь свойства length, даже в этом случае вы хотите продолжить работу с потоком.Если это так, тогда вы можете добавить простой try catch в do()

  from([1, 2])
  .do(a => {
      try {
        console.log(a);
        let d:string = null;
        let r = d.length;
      } catch(e) {
        console.log(`catched under do`, e)
      }
  })
  .catch(() => {
      console.log("error catched");
      return Observable.of('catched');
  })
  .subscribe(data => {
    console.log(`data`, data)
  },(e) => {
      console.log(`error`, e)
  });

Вот пример: https://stackblitz.com/edit/angular-bcrava?file=src/app/app.component.ts

Answer 2

Вы получите вывод

1 
error catched

потому что первое значение выдает ошибку в tap, которая распространяется по каналу onError, который останавливает последовательность. С catch вы ловите эту ошибку и переходите к следующей последовательности (never), но первая последовательность (of(1,2)) все еще останавливается. Поэтому после 1 выдает ошибку 2 никогда не обрабатывается tap.

После того, как вы вернете never в catch, значение не будет отправлено, и наблюдаемое никогда не завершится. Если вы добавите журналы внутри subscribe для обратных вызовов next, error и complete, вы увидите, что они никогда не выполняются.