angular - rxjs/可观察到的 : Execute functions one by one and pass parameters of functions as result of previous function

Question

我有一个包含服务队列的数组，我需要一个接一个地运行这些服务，并将第一个函数的结果作为第二个函数的参数传递。服务可以是同步的和异步的。服务的结果可以是解决、拒绝或一些结果值。如果队列中的一个服务失败，则所有队列都应该失败，并且不运行队列中的下一个服务。

目前，我有一个 Promises 实现

let collection = [
 { name: 'calcOnServer', parameters: {} },
 { name: 'calc', parameters: {} }
];

return collection.reduce((currentFunction, nextFunction, index) => {
  return currentFunction.then(() => {
    let result = runFunction(nextFunction.name, nextFunction.parameters);
    // runFunction is some method that can get class with name from first parameter and execute for it method 'run' with parameters from second parameter


    if (result === undefined) {
      result = Promise.resolve();
    }

    if (!result.then) {
      if (Boolean(result)) {
        result = Promise.resolve(result);
      } else {
        result = Promise.reject();
      }
    }

    return result.then((result) => {
      collection[index + 1].parameters = result;
    });
  });
}, Promise.resolve())

目前服务可以看起来像

class calcOnServer {
  run({param1, param2}) {
    return new Promise((resolve, reject) => {
        // some async operation
       .then(resolve, reject);
    }
  }
}

class calc {
  run({param1, param2}) {
    if (typeof param1 === 'number' && typeof param2 === 'number') {
      return param1 + param2
    } else {
      return Promise.reject();
    }
  }
}

我需要用 RxJS/Observables 重写这个逻辑。

Answer 1

我不完全确定你的代码应该做什么，但一般来说，当你想迭代一个数组并对所有数组运行相同的异步操作时，你可以使用 mergeScan 运算符.它还接受一个可选的 concurrency 参数，告诉它你想运行多少个并行操作:

import { of, from } from 'rxjs'; 
import { map, mergeScan, delay } from 'rxjs/operators';

const collection = [1, 2, 3, 4, 5];

const source = from(collection).pipe(
  mergeScan((acc, value) => {
    console.log(acc, value);
    return of(value).pipe( // Your async operation will be here
      delay(1000),
      map(response => `response-${response}`)
    );
  }, null, 1), // `1` will make `mergeScan` process items one by one
);

source.subscribe();

此示例将打印以下输出。请注意，每个日志都包含上一次调用的响应和当前值:

null 1
response-1 2
response-2 3
response-3 4
response-4 5

现场演示:https://stackblitz.com/edit/rxjs-3jsae2