javascript - 如何在包含同步和异步函数的 Observables 的 RxJS 中创建合适的队列

Question

我的应用程序从服务器接收响应。这些响应可能会被单个或多个接收，但它们将始终按照它们应该被单独处理和呈现的顺序接收。

收到响应后，需要按顺序发生以下事情:

1. 呈现并显示输入指示器(同步)
2. 响应数据被解析并转换为正确的格式(同步)
3. 运行人为延迟以模拟“真人”打字(异步)
4. 应删除/隐藏输入指示器(同步)
5. 将格式化的响应呈现到屏幕(同步)

现在，所有这些步骤都需要按顺序进行，以便收到每个响应，也按顺序进行。这样做的最终结果应该是打字指示符显示随机时间，然后消失，然后显示响应，然后指示符再次显示随机时间，然后消失，然后显示下一个响应，等

我已经尝试了多种技术来使用 RxJS 来实现这一点，但我正在努力让它发挥作用。我发现无论采用何种方法，响应(添加到类似队列的结构和/或按顺序执行)都会同时触发事件链。我无法弄清楚如何按顺序执行事件链并在下一个事件开始之前等待每个先前的事件链完成。

我尝试使用包含 .tap 和 .concatMap 函数的 .pipe 链解决方案，并从 new Subject() 并向其中添加新项目以使用 next() 执行。这些都不能解决我的问题。

核心构造必须是某种队列，我可以继续向其添加项目，并且当队列完成/为空时它们将自动执行。

Answer 1

听起来像是 concatMap operator 的工作.在下面的示例中更改回调以更新您的 UI/状态，它应该可以完成工作:

const { of, concat } = rxjs; // = require("rxjs")
const { delay, concatMap } = rxjs.operators; // = require("rxjs/operators")

const responses$ = of(1, 2, 3, 4, 5);

const events$ = responses$.pipe(
  // concatMap to process items sequantually
  concatMap(e => {
    const typing$ = of({ type: "typing" });
    // add delay before receiving new message
    const message$ = of({ type: "new message", message: e }).pipe(delay(3000));
    
    // add delay before each response
    return concat(typing$, message$).pipe(delay(2000));
  })
);

events$.subscribe((e) => {
  console.log(e);
});

<script src="https://unpkg.com/rxjs@6.5.2/bundles/rxjs.umd.min.js"></script>