Rxjs bindNodeCallback 不能与 swithmap 和 flatmap 一起正常工作

Question

我正在尝试使用 bindNodeCallback 和 fs readdir, stat 读取文件。

这是我的代码：

import { readdir, stat, Stats } from "fs";
import { bindNodeCallback, Observable, of } from "rxjs";
import { catchError, filter, flatMap, map, mergeMap, switchMap, tap } from 'rxjs/operators';

const readDirBindCallback: (path: string) => Observable<string[]> = bindNodeCallback(readdir);
const fileStateBindCallback: (path: string) => Observable<Stats> = bindNodeCallback(stat);

readDirBindCallback('/Users/.../WebstormProjects/copy')
    .pipe(
        flatMap(x => x),
        tap(console.log),
        switchMap(status => {
            console.log(status);
            return fileStateBindCallback('/Users/.../WebstormProjects/copy/' + status);
        })
    )
    .subscribe(result => {
        console.log(result)
    });

“switchMap”已正确调用多次。

问题是 fileStateBindCallback 只被调用了一次，最终的订阅只记录了一次。

为什么会这样？ 如果我在 switchMap 块中手动订阅 fileStateBindCallback。它会按我的预期运行。但这不是一个好的做法，也不符合我的要求。

Answer 1

我认为问题在于switchMap 运算符。

switchMap 一次只能有一个活动的内部可观察对象。如果有一个外部值进来并且有一个活动的内部 obs.，它将被取消订阅，并且将根据 新到达的值 和 创建一个新的值 >提供功能到switchMap。

readdir 将返回一个array。

flatMap(arr => arr) 将简单地分别和同步发出数组的项。

假设您的目录有N 项。

flatMap 将一一发出这些项目。 所以item 1 被传递，switchMap 将创建一个内部 obs (fileStateBindCallback)，这涉及到异步操作。
然后，item 2 被发送，但由于这是同步发生的，switchMap 处理的当前内部 observable（由于item 1 而创建的）将被取消订阅，并且将为item 2 创建一个新的。

以此类推，直到item N 终于到来。请记住，数组的项目是同步发出的。 item N-1 的内心世界。将被取消订阅并为item N 创建一个新的。但由于N 是数组中的最后一项，它不能被任何后续的发射打断，这意味着它的内部可观察对象将有时间发射和完成。

您可能正在寻找concatMap。

当Nth 项到达时，它将等待由于N-1th 项而创建的当前内部可观察对象完成，然后再创建一个新一个基于Nth 项目。

因此，将 switchMap 替换为 concatMap 应该可以完成这项工作。如果顺序不重要，可以选择mergeMap。