在管道内使用 concatMap 两次的更好替代方案？

Question

我想在另一个承诺（承诺 1）解析后递归执行一个预定义的承诺（承诺 2），以查看我与外围设备的连接何时返回 false，然后我可以执行其他操作。我希望能够使用可观察量来做到这一点。这是我到目前为止编写的代码，有两个问题：

1）是否有更好的替代方案可以在 reprovision() 函数中的代码中使用 concatMap 两次 ，

以及

2）是否有更好的方法来实现我的代码来执行我上面描述的操作？

下面是使用 observables 递归执行 Promise 1（仅一次）和 Promise 2，然后对结果执行某些操作的代码：

reprovision(){
    this.bleMessage = this.prepareMessageService.prepareMessage(
      MessageCategory.customAction, CustomActionConstants.reboot, this.customPayload
    );

    from(this.bleService.bleWrite('custom', this.bleMessage)).pipe(
      concatMap(x => interval(1000)),
      concatMap(x => this.bleService.checkBleConnection1(this.deviceId)),
      takeWhile(x => x, true)).subscribe(
        resp => console.log('still connected to peripheral'),
        err =>  {
          console.log('disconnected from peripheral, do something else');
        }
      );
}

这是 Promise 2 (this.bleService.checkBleConnection1) 的代码，我希望递归执行它，直到外围设备断开连接从我的移动设备。这个承诺背后的想法是，只要我的移动设备仍然连接到外围设备，它就应该解决，并在连接失败时抛出错误。

checkBleConnection1(deviceId: string){
  return BleClient.getConnectedDevices([]) //BleClient.getConnectedDevices([]) returns a promise
  .then(resp => {
    if (!resp.some(elem => deviceId === elem.deviceId)){
      throw new Error('Peripheral connection failed');
    }
    return true;
  })
  .catch(error => {
    throw error;
  });
}

Answer 1

从您的帖子中我了解到，

• 如果找到 设备，您希望通过无限调用 this.bleService.checkBleConnection1(this.deviceId) 以 1 秒的间隔进行递归检查
• 。如果找不到设备，则需要检查更多内容

因此，对于递归调用，Rxjs 中有一个名为 expand 链接

要实现 1 秒间隔添加 delay(1000) 在返回递归调用时

所以这是我的版本，使用 expand 代替 concatMap(x =>; Interval(1000))

使用 expand 不需要使用 takeWhile，所以我将其删除

reprovision(){
  this.bleMessage = this.prepareMessageService.prepareMessage(
    MessageCategory.customAction, CustomActionConstants.reboot, this.customPayload
  );

  from(this.bleService.bleWrite('custom', this.bleMessage))
    .pipe(
      expand((x) => from(this.bleService.checkBleConnection1(this.deviceId)).pipe(delay(1000)))
    )
    .subscribe({
      next: (resp) => console.log('still connected to peripheral'),
      error: (err) =>
        console.error('disconnected from peripheral, do something else'),
      complete: () => console.log('Completed !'),
    });
}

Answer 2

我相信这里的目标是在完成第一个承诺后每 1 秒检查一次连接（旁注：这称为轮询，而不是递归，如果您调用 reprovision() 从内部 reprovision() 或类似的东西，您可以递归轮询，但您不在这里并且通常不想这样做，除非必须这样做）。

你不能真正摆脱第二个 concatMap 因为你必须根据时间间隔切换到它，你可以像这样分离和清理流：

const pollConnection$ = interval(1000).pipe(
  concatMap(i => this.bleService.checkBleConnection1(this.deviceId)),
  takeWhile(x => x, true)
);

from(this.bleService.bleWrite('custom', this.bleMessage)).pipe(
  concatMap(x => pollConnection$),
).subscribe(
    resp => console.log('still connected to peripheral'),
    err =>  {
      console.log('disconnected from peripheral, do something else');
    }
  );

但你需要小心使用像 concatMap 这样的运算符，因为它会产生背压，这对应用程序不利。如果这些请求的响应时间可能超过 1 秒，那么您将构建一个请求队列，这可能会降低应用程序的性能。更安全的替代方案是 switchMap 和 exhaustMap，具体取决于所需的行为