Исключение в наблюдаемой трубе подавлено

Question

Предположим, я хочу, чтобы наблюдаемая периодически генерировала значение до тех пор, пока не появится другая наблюдаемая. Так что я могу использовать timer и takeUntil, чтобы добиться этого. Но затем я хочу обработать каждое испущенное значение и прекратить (ошибку) испускание, когда какое-либо условие становится истинным. Поэтому я пишу следующий фрагмент кода:

const { timer, Subject } = 'rxjs';
const { takeUntil, map } = 'rxjs/operators';

const s = new Subject();
let count = 0;
function processItem(item) {
  count++;
  return count < 3;
}

const value = "MyValue";

timer(0, 1000).pipe(
      takeUntil(s),
      map(() => {
        if (processItem(value)) {
          console.log(`Processing done`);
          return value;
        } else {
          s.next(true);
          s.complete();
          console.error(`Processing error`);
          throw new Error(`Stop pipe`);
        }
      }),
    )

Детская площадка

Но вместо того, чтобы получить ошибку, я завершил свой Observable. Только если я закомментирую оператор takeUntil(s), я получаю ошибку. Похоже, когда оператор pipe завершает работу, его значение не выдается сразу, а запоминается и выдается в конце следующей «итерации» канала, затем заменяется новым результатом и так далее. И в моей ситуации следующая итерация, когда должна появиться ошибка, предотвращается takeUntil. Вопрос в том, правильно ли я с этим предположением, и если да, то почему rxjs разработан таким образом?

Answer 1

Прежде всего, каждая цепочка Rx может отправлять одно error или одно complete уведомление, но не оба одновременно. (См. http://reactivex.io/documentation/contract.html раздел «Контракт, регулирующий уведомления» ).

takeUntil оператор выдает complete, когда его уведомление Наблюдаемое (s в вашем случае) выдаетлюбое next уведомление. Это означает, что когда s испускает, цепочка будет завершена, и вы никогда не получите никаких дальнейших error уведомлений.

Последнее и, вероятно, самое запутанное, что все в RxJS происходит синхронно, если вы не работаете со временем (например, оператор delay) или вы специально используете оператор observeOn с асинхронным планировщиком. Поэтому, когда вы вызываете s.next(true) внутри map, это next уведомление немедленно распространяется на takeUntil, что завершает цепочку, и, как я уже упоминал выше, вы можете получить одно error или одно complete уведомление, но никогда оба.

Похоже, вам даже не нужно использовать takeUntil, потому что если вы выбросите ошибку внутри map, она автоматически обернута и отправлена ​​далее как error уведомление ( Как броситьошибка оператора карты RxJS (угловая) ) и цепочка удаляется автоматически, поэтому нет смысла пытаться завершить ее после этого с помощью takeUntil.