从嵌套的 observables 中进行 Angular RXJS 轮询

Question

我有一个服务，解析器使用它来生成和返回报告。

服务中的初始获取调用一个 REST 端点 /report，它会在服务器上启动一个工作作业，因为报告是处理器密集型的并且需要 30 多秒才能运行。 report 端点返回工作任务的 ID。

然后，我需要使用作业的相关 ID 轮询工作人员作业 REST 端点 /job/job_id。我继续轮询，直到它返回“完成”状态，并包含完成的报告。

这个最终输出然后从服务返回，解析器使用它。

我无法通过投票来解决这个问题。我将对初始报告端点的响应通过管道传输到 switchMap 中，然后使用间隔每隔 500 毫秒重复轮询/job/job_id 端点。然后我尝试切换民意调查响应并在完成时返回。这是我第一次使用 switchMap 和 polling，所以我不确定我是否正确使用它。

这是我最近尝试的代码：

getDepartmentReport() {

  return this.http
    .get<any>(reportUrl, this.getAuthOptions(true))
    .pipe(switchMap(initialResponse => {

      interval(500).pipe(
        switchMap(() => {
          return this.http.get<any>(workerUrl + initialResponse.id, this.getAuthOptions(true))
            .pipe(
              switchMap(pollResponse => {
                if(pollResponse.state === 'completed') {
                  return pollResponse;
                }
            })
      }));
  }));
}

这实际上不会编译。它给出了以下错误：

Argument of type '(initialResponse: any) => void' is not assignable to parameter of type '(value: any, index: number) => ObservableInput<any>'.
  Type 'void' is not assignable to type 'ObservableInput<any>'.

56         .pipe(switchMap(initialResponse => {

我认为这种情况正在发生，因为在不完整的轮询响应中，没有返回语句来处理这种情况，并且正在返回一个 void。

有人有什么想法吗？我被难住了。

Answer 1

这是一个有趣的问题。

您收到该错误是因为switchMap 必须返回一个Observable。在你的代码中，你没有返回任何东西，你只是开始一个间隔。

您还必须告知停止轮询的时间间隔。这可以在takeWhile 运算符的帮助下实现。为了进一步区分事物，我创建了一个自定义运算符，将在其中进行轮询。 这样做的话，你也可以在其他地方复用这个操作符。

这是我的方法：

// ===== Server =====

let crtReportId = 1;
let crtReportStatus: { status: string, id: number };

const getReportFromBE = () => {
  let initialId = crtReportId;

  crtReportStatus = { status: 'pending', id: initialId };

  // It takes some time...
  timer(2000)
    .subscribe(() => crtReportStatus = { status: 'completed', id: initialId })

  return of(crtReportId++);
}

const getWorkerStatus = id => of(crtReportStatus);

// ===== Client =====

type CustomPollOperator = (data: any, cond: (d) => boolean, ms: number) => Observable<any>

const pollFor: CustomPollOperator = (data, cond, ms) => {
  let shouldPoll = true;

  return interval(ms)
    .pipe(
      tap(() => console.warn('pooling', shouldPoll)),
      takeWhile(() => shouldPoll),
      switchMap(() => getWorkerStatus(data)),
      tap(res => {
        if (cond(res)) {
          shouldPoll = false;
        }
      })
    )
}

const isWorkerCompleted = w => w.status === 'completed';

const getReports = () => {
  return getReportFromBE()
    .pipe(
      switchMap(workerId => pollFor(workerId,isWorkerCompleted, 200))
    )
}

getReports().subscribe((res) => console.log('result', res))

StackBlitz.