rxjs 如何优雅的处理轮询任务？

需求：

假设异步请求返回有一个状态，值为 pending 或者 success
异步请求如果返回 pending，则等待一秒后重新发送这个异步请求，直到返回 success
最终使用 subscribe 只返回最终结果，中间过程的输出不走 subscribe

自己的实现

自己尝试实现了，但是结果不是自己想要的，而且如果多个异步轮询（ A 轮询 -> B 轮询 -> C 轮询），可能就会使代码变得嵌套层级多，以下是实现的具体代码，有大佬指教指教么？

轮询代码

import { from, Observable, Subscriber } from 'rxjs';
import { delay, mapTo, repeatWhen } from 'rxjs/operators';

interface RetryOptions<T = any, P = any> {
  try: (tryRequest: P) => Promise<T>;
  tryRequest: P;
  retryUntil: (response: T) => boolean;
  maxTimes?: number;
  tick?: number;
}

export const polling = <T = any, P = any>(options: RetryOptions<T, P>) => {
  options = Object.assign(
    {
      maxTimes: 20,
      tick: 1000
    },
    options
  );
  let result = null;

  const notifier = () => {
    // 计数最大尝试次数
    let count = 0;
    const loop = (producer: Subscriber<any>) => {
      // 超过最大次数强制退出轮询
      if (count >= options.maxTimes) {
        producer.complete();
      } else {
        options
          .try(options.tryRequest)
          .then(res => {
            producer.next(count++);
            // 满足条件则退出轮询
            if (options.retryUntil(res)) {
              producer.complete();
            } else {
            // 不满足条件则继续轮询
              loop(producer);
            }
            // 保存请求结果
            result = res;
          })
          .catch(err => {
            producer.error(err);
          });
      }
    };
    return new Observable(producer => {
      loop(producer);
    });
  };

  return from([0]).pipe(
    delay(options.tick),
    // 当满足条件是，进行一下轮轮询
    repeatWhen(notifier),
    // 转换结果
    mapTo(() => result)
  );
};

测试用例

import { polling } from './polling';

let count = 0;

const mockRequest = (): Promise<string> => {
  return new Promise((resolve, reject) => {
    setTimeout(() => {
      if (count < 6) {
        resolve('pending');
      } else {
        resolve('finish');
      }
      count++;
    }, 1000);
  });
};

polling<string, number>({
  try: mockRequest,
  tryRequest: count,
  retryUntil: res => {
    return res === 'finish';
  }
}).subscribe((response) => {
  const result = response();
  console.log(result);
  if (result === 'finish') {
    console.log('轮询结束');
  }
  // 这个轮询结束后应该怎么继续轮询比较好？
  // 继续在这里 polling 下一个轮询吗？容易回调地狱啊
});

结果

null
pending
pending
pending
pending
pending
pending
finish // 上面的都不输出，只输出最后一个结果，因为上面的我并不关注
轮询结束

wawaforya

2019-08-23 18:34:49 +08:00

献丑了，有什么错误请轻拍😂

``` typescript
import { Observable, of, race, timer } from 'rxjs';
import { concatMapTo, skipWhile, take, tap } from 'rxjs/operators';

type Result = 'pending' | 'success';
const getResult: () => Observable<Result> = () => of<Result>('success').pipe(tap(() => console.log('Requested.'))); // 向接口请求数据的函数

const limit = 20;
const schedule = timer(0, 1000);
const requestSource = schedule.pipe(concatMapTo(getResult()), skipWhile(result => result === 'pending'));
const upperBound = schedule.pipe(skipWhile(value => value < limit));
race([requestSource, upperBound]).pipe(
take(1)
).subscribe(
result => console.log(result), // 这里需要判断类型，如果是数字，说明 20 次了都还没有 success
error => console.error(error),
() => console.log('Completed')
);
```