rxjs 在每次迭代后更新数组的可观察值

Question

我有一个 Observable，其中包含一个数组，我在 UI 中使用该数组与异步管道来显示合约表。该表中的一列显示页面加载时“未处理”的状态值。

contracts$: Observable<ContractDto[]>

在视图初始化时，我从像

this.contracts$ = this.api.getContracts()

这样的 API 加载合约

效果很好，现在解决我的问题：

通过单击按钮，用户可以启动一个在所有合约上运行的进程，以执行每个合约的操作。每个操作都包含一个 API 调用，可能需要 1-3 秒。

现在我想更新视图以向用户显示哪些合同已被处理，以便他获得有关处理进度的指示器。因此，我想迭代该可观察对象中的合约，并在执行操作时将状态字段更新为“已处理”。

但是我找不到一种方法来迭代可观察的合约并在每次迭代时更新 this.contracts$ 。我唯一能做的就是一一迭代和处理，但只有在处理完所有操作后才更新this.contracts$。

我当前的代码如下所示：

// this only gets updated once all contracts have been processed
this.contracts$ = this.contracts$
            .pipe(
                // flatten array of contracts
                concatMap(contract => contract),

                // executing the process one by one; returning the updated contract
                concatMap(contract => this.deploy(contract)),

                // without toArray() i get an error on this.contracts$ assignment because 
                // ContractDto can not be assigned to ContractDto[] - which is correct.
                // and here is my missing link because I want to update this.contracts$ after EACH iteration.
                toArray(),
            )

非常感谢您的宝贵时间

Answer 1

如果你想在每次迭代后更新contracts$ observable，你可以使用scan操作符和concatMap。扫描运算符允许您随着时间的推移累积值，这对于增量更新合约数组非常有用。