angular - Concat 可观察对象

Question

我有一些功能可以与数据库一起使用:

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

在另一个函数的循环内，我需要按顺序运行这些可观察值。我这样做:

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。您可以在可观察对象上执行此运算符，它会做的是。

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 创建了一个可观察的 storageIds$。然后我对这个可观察的对象执行 concatMap。我只是调用返回一个可观察对象的 setupData 方法，并立即返回它。这将为您提供预期的行为。

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