如何通过 rxjs observable 设置当前索引或向后移动？答案

【问题标题】：How to set current index of or move backwards though an rxjs observable?如何通过 rxjs observable 设置当前索引或向后移动？
【发布时间】：2020-09-22 14:10:56
【问题描述】：

我正在使用 rxjs observable 对我的应用程序中发生的一系列步骤进行建模。

伪代码

const steps = [{id: "step1"}, {id: "step2"}, {id: "step3"}]

//... then at some point later do things based on the steps
from(steps).pipe(
    concatMap((step) => {
        ... do stuff with each step, perhaps including needing to go back to previous step
    })
)

鉴于这种模式，如果某些条件需要，我该如何模拟“返回”？

例如从 step1 -> step2 -> step3 -> step2 -> step3

据我所知，rxjs 似乎是非常严格的单向线性的。我还没有找到一种方法来设置可观察对象的位置/索引（如果这是将其概念化的正确方法）。

抱歉，如果我遗漏了一些明显的东西，但这似乎是从数组创建的 observables 的一个相当常见的用例。

【问题讨论】：

鉴于 Observables 是在特定时间点发生的事件流，可能很难返回 ;-) 猜猜我们需要以比光速更快的速度旅行
您也许可以在某种列表中“保存”步骤。
基本上你无法知道流是否来自其他来源的数组。您始终可以使用 repeat 运算符重新订阅源。如果天气很冷，就像您的情况一样，您将再次收到所有通知。如果天气很热，您需要使用 shareReplay 之类的东西来重播过去的事件。

标签： rxjs rxjs6 rxjs-pipeable-operators rxjs-observables

【解决方案1】：

我想出了一种方法来做到这一点，但我不确定这是否是“正确”的模式。

我创建了一个主题并让它向自身发出值以控制我的序列中的哪个项目接下来出现。看起来像这样：

const steps = [{id: "step1"}, {id: "step2"}, {id: "step3"}, {id: "step4"}]

const runMySequence = async () => {

    let currentStep = steps[0]
    const sub = new Subject<typeof steps[number]>()

    sub.pipe(
         tap((step) => currentStep = step),
         concatMap(....process the step, do all sorts of stuff),
         tap((step) => {
             //Here's where we decide which step to execute next
             const idx = steps.indexOf(currentStep)
             if (idx !== steps.length - 1) {
                //Based on whatever condition we could decide to "go back" to a previous step here if we wanted
                sub.next(steps[idx + 1])
             } else {
                 //All out of steps, complete the subject
                 sub.complete()
             }
          })).subscribe()

    //kick off the sequence
    sub.next(steps[0])

    //and wait for it to finish
    await sub.toPromise()

}

这似乎是实现这一目标的一种相当干净的方法。对为什么这种模式可能存在问题或更好的方法有什么想法吗？

【讨论】：