RxJS Observable: повторите с использованием count, а затем с помощью уведомителя

Question

У меня есть Observable, который испускает Either = Success | Failure:

import { Observable } from 'rxjs';

type Success = { type: 'success' };
type Failure = { type: 'failure' };

type Either = Success | Failure;

const either$ = new Observable<Either>(observer => {
    console.log('subscribe');
    observer.next({ type: 'failure' });
    observer.complete();
    return () => {
        console.log('unsubscribe');
    };
});

Я хочу, чтобы пользователь мог «повторить» наблюдаемое, когда «Наблюдаемое» завершится, и последнее значение будет Failure.

(Операторы retry{,When} здесь не помогают, потому что они работают с ошибками на канале error. По этой причине я считаю, что вместо этого мы должны думать в терминах repeat.)

Я хочу:

• Повторяйте Наблюдаемое n раз, пока последнее значение не станет Failure.
• Затем разрешите пользователю повторить вручную. Когда изменится наблюдаемое уведомление о повторных событиях (repeat$), повторите наблюдаемое снова.

Например:

// subscribe
// next { type: 'failure' }
// unsubscribe

// retry 2 times:

// subscribe
// next { type: 'failure' }
// unsubscribe

// subscribe
// next { type: 'failure' }
// unsubscribe

// now, wait for repeat notifications…
// on retry notification:

// subscribe
// next { type: 'failure' }
// unsubscribe

Answer 1

Я не мог придумать что-нибудь попроще, но код делает то, что вы хотите.

См. https://stackblitz.com/edit/typescript-yqcejk

defer(() => {
   let retries = 0;

   const source = new BehaviorSubject(null);

   return merge(source, repeat$.pipe(filter(() => retries <= MAX_RETRIES)))
       .pipe(
           concatMapTo(either$),
           tap(value => {
               const action = value as Either;
               if (action.type === 'failure') {
                   if (retries < MAX_RETRIES) {
                       retries += 1;
                       source.next(null);
                   }
               } else {
                   retries = 0;
               }
           })
       )
}).subscribe(console.log);

Мне пришлось вручную считать повторы.

Код имеет два источника событий source для автоматических повторных попыток и repeat$ для пользовательских повторных попыток. Все события отображаются на either$ с использованием concatMapTo. В качестве побочного эффекта мы либо next() повторяем попытку, либо ничего не делаем, ожидая повторной попытки пользователя.

Пользовательские попытки подавляются с помощью filter(() => retries >= MAX_RETRIES) до тех пор, пока не будет достигнут MAX_RETRIES.