RxJS Observable：重复使用计数，然后使用通知器

Question

我有一个发出 Either = Success | Failure 的 Observable：

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');
    };
});

我想让用户在 Observable 完成并且最后一个值为 Failure 时“重试”observable。

（retry{,When} 运算符在这里没有帮助，因为它们在 error 频道上处理错误。因此，我认为我们应该改为考虑 repeat。）

我想：

• 重复 Observable n 次，直到最后一个值不是 Failure。
• 然后，允许用户手动重复。当重复通知 observable (repeat$) 发出时，再次重复 observable。

例如：

// 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$ 用于用户重试。所有事件都使用concatMapTo 映射到either$。作为副作用，我们要么next() 重试，要么什么都不做，等待用户重试。

使用filter(() => retries >= MAX_RETRIES) 禁止用户重试，直到达到MAX_RETRIES 计数。