RxJS 节点等待 observable 完成，然后再按顺序订阅下一个

Question

我正在尝试使用从谷歌表格中抓取的值填充我的数据库。 hasuraHelper.insert() 将 HTTP 发布到我的数据库。如何使 insert() 仅在上一个 api 调用返回后才被调用？我认为 concatMap 可以解决问题，但它似乎仍然急切地订阅所有排放。

当前：请求 1 -> 请求 2 -> ... -> 请求 1 完成 -> 请求 2 完成 -> ...

我想要什么：请求 1 -> 请求 1 完成 -> 请求 2 -> 请求 2 完成 -> ...

下面是相关代码：

sheetsHelper.authToken()
    .flatMap(sheetsHelper.get)
    .flatMap(response => response.values) //values is a 2d array
    .map(row => {
        console.log(row[0]);
        const flights = [];
        //add 100-500 objects to this array
        return flights;
    })
    .flatMap(flights => {
        const arrayOfFlightArrays = [];
        //max batch size of 50
        while (flights.length) {
            const flightArr = flights.splice(0, 50);
            arrayOfFlightArrays.push(flightArr);
        }
        return Rx.Observable.from(arrayOfFlightArrays);
    })
    .concatMap(flights => hasuraHelper.insert(flights) //insert flights into table
        .retry()) 
    .map(response => response.data)
    .subscribe(console.log, console.error);

这是插入的样子：

exports.insert = function (objects) {
    return Rx.Observable.fromPromise(axios({
        method: 'post',
        url: '/',
        data: {
            type: "insert",
            args: {
                table: "flights",
                objects: objects
            }
        }
    }));
};

Answer 1

您的插入函数并不懒惰。你在 Observable 中封装了一个正在执行的 Promise，但它已经在执行了。

要获得一个承诺变得懒惰并在订阅时开始处理，您需要将其包装在Rx.Observable.defer(() => Rx.Observable.fromPromise(axios(/* ... */))