如何设置rxjs可观察到的当前索引或向后移动？

Question

我正在使用可观察到的rxjs来建模我的应用程序中发生的一系列步骤。

伪代码

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
    })
)

鉴于这种模式，如果有必要，我将如何建模“返回”？

例如从步骤1->步骤2->步骤3->步骤2->步骤3

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

很抱歉，我缺少明显的东西，但是对于从数组创建的可观察对象来说，这似乎是相当普遍的用例。

Answer 1

我想出了一种方法来执行此操作，但是不确定这是否是“正确的”模式。

我创建了一个主题，并使其本身发出值以控制序列中的下一个项目。看起来像这样：

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

const runMySequence = async () => {

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

    sub.pipe(
         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(self.currentStep)
             if (idx !== self.steps.length - 1) {
                //Based on whatever condition we could decide to "go back" to a previous step here if we wanted
                sub.next(self.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()

}

这似乎是一种很干净的方法。是否对为什么这种模式可能会有问题或更好的方法有任何想法？