Concat 可观察对象

Question

我有一些函数可以使用 DB：

setupData(param) {
    return Observable.create((observer) => {
        this.db.executeSql('SELECT .... ?', param).then(() => {
            console.log('DB QUERY DONE');
            observer.complete();
        }, (error) => {
            observer.error(error);
        });
    });
}

在另一个函数的循环中，我需要按顺序运行这些 observables。我是这样做的：

processData() {

    ...

    let observers = [];
    storage.forEach((data) => {
        observers.push(this.setupData(data));
    });

    Observable.concat(observers).subscribe(() => {
        console.log('NEXT');
    }, () => {
        console.log('ERROR');
    }, () => {
        console.log('COMPLETE');
    });
}

所以我的输出：

NEXT
NEXT
NEXT
...
NEXT
NEXT
COMPLETE

但我从未见过“DB QUERY DONE”。每个可观察对象都没有订阅。如果我用 forkJoin 替换 concat - 我看到我期望，但我需要按顺序运行，而不是并行运行......有什么解决方案吗？

Answer 1

您真正需要的是 concatMap。你可以在一个 observable 上执行这个操作符，它会做什么。

1. 取下一个值。

2. 对该值执行一个函数，该函数应返回一个可观察的值。

3. 开始监听这个 observable 并将值从这个 observable 发送到下一个。

4. 一旦这个 observable 停止发射，从源 observable 获取下一个值并重复。

代码看起来像这样：

const processData = () => {
    let storageIds$ = Rx.Observable.from(["1", "2", "3"]);

    storageIds$.concatMap((val) => {
      return setupData(val);
    }).subscribe(() => {
        console.log('NEXT');
    }, () => {
        console.log('ERROR');
    }, () => {
        console.log('COMPLETE');
    });
}

你可以看到我已经从数字 1、2、3 创建了一个 observable storageIds$。然后我在这个 observable 上执行一个 concatMap。我只是调用 setupData 方法，它返回一个 observable，并立即返回。这将为您提供预期的行为。

完整的 jsbin 示例可以在这里找到： http://jsbin.com/katame/edit?js,console