RxJS - 控制多个并行执行

Question

我有一个场景，我必须控制多个 PTZ 摄像机，每个摄像机拍摄多个角度的照片。因此，例如:
相机 A 将拍摄角度 A1、A2、A3
摄像机 B 将拍摄角度 B1、B2、B3、B4

移动相机以指向正确的角度、捕获图像并上传图像，都是返回 Promise 的异步函数。
moveCamera(角度)
captureImage()
上传图片()

摄像机必须并行操作，但每个摄像机拍摄的角度必须按顺序完成。

不知何故，我觉得 RxJS 可以轻松解决这个问题，但我正在努力将它们拼凑在一起。我能做的最好的就是像下面的解决方案，它以某种方式使相机按顺序相互处理。请注意，我使用的是 redux-observable，下面的代码是我在普通 RxJS 中能找到的最好的代码。请原谅我的 RxJS。

const angles = {
  'Camera A': [ 'A1', 'A2', 'A3' ],
  'Camera B': [ 'B1', 'B2', 'B3', 'B4' ],
}
const cameras = of( [ 'Camera A', 'Camera B' ] );
const cameraRun = cameras.pipe(
  mergeMap( camera => {
    // in redux-observable, I could return an array here
    return of( angles[ camera ] );
  } )
);
cameraRun.pipe(
  concatMap( angle => {
    return moveCamera( angle )
      .then( () => captureImage() )
      .then( () => uploadImage() )
      .then( () => console.log( 'Image success' ) );
  } )
)

对于那些了解 redux-observable 的人，我有 3 个史诗:
RUN_CAMERA_SET_ROUTINE - 运行 mergeMap
内的所有摄像机RUN_CAMERA_ROUTINE - 运行mergeMap内每个摄像机的所有角度
CAPTURE_IMAGE - 在一个 concatMap 内运行上面的异步函数

我最初的想法是由于 mergeMap 生成流，CAPTURE_IMAGE 将被“分组”，但我错了。看来 CAPTURE_IMAGE 仍然是一个流，对所有摄像机的每个角度进行排队。

任何指示都会非常有帮助。

Answer 1

您的问题归结为按顺序并行执行一些 Observables，并从 Promises 创建 Observables。

1. 要并行执行多个 Observable，请使用:

• forkJoin ，如果您只想在所有相机操作完成后发出最终的 Observable
• merge ，如果您希望每次单个相机 Action 成功时发出最终的 Observable

使用concat按顺序执行多个 Observable。

使用defer从 Promise 创建 Observable 但不立即执行 Promise。

然后你必须

• 构造一个要按顺序执行的 Observable 数组。
  (单个摄像机每个角度的 Action )
• 构造一个要并行执行的 Observable 数组。
  (每个摄像机的摄像机 Action )。

这可能是纯 RxJS 中的代码

import { concat, forkJoin, merge, defer } from 'rxjs';

const cameras = ['Camera A', 'Camera B'];
const cameraAngles = { 
  'Camera A': ['A1', 'A2', 'A3'], 
  'Camera B': ['B1', 'B2', 'B3', 'B4'] 
}

// Performs a camera action consisting of multiple parts. Returns a Promise.
// camera: e.g. 'Camera A', angle: e.g. 'A1'
const doCameraAction = (camera, angle) => moveCamera(angle)
  .then(() => captureImage())
  .then(() => uploadImage())
  .then(() => console.log('Image success'));

// Creates an Observables that executes multiple camera actions in sequence.
// camera: e.g. 'Camera A', angles: e.g. ['A1', 'A2', 'A3']
const getCameraActionSequence$ = (camera, angles) => concat(
  // the array of Observables we want to execute in sequence
  ...angles.map(angle => defer(() => doCameraAction(camera, angle)))
);

// An Observable that will execute multiple camera action sequences in in parallel
const multiCameraActions$ = forkJoin(
  // the array of Observables we want to execute in parallel
  cameras.map(camera => getCameraActionSequence$(camera, cameraAngles[camera]))
);

https://stackblitz.com/edit/rxjs-gj1dny?file=index.ts