使用生成器/迭代器包装结果集回调函数

Question

我正在努力将基于回调的旧 API 转换为异步库。但我就是无法将“结果集”作为生成器（节点 10.x）工作。

原来的 API 是这样工作的：

api.prepare((err, rs) => {
    rs.fetchRows(
        (err, row) => {
            // this callback is called as many times as rows exist
            console.log("here's a row:", row);
        },
        () => {
            console.log("we're done, data exausted");
        }
    );
});

但这是我想使用它的方式：

const wrapped = new ApiWrapper(api);
const rs = await wrapped.prepare({});
for (let row of rs.rows()) {
    console.log("here's a row:", row);
}

let row;
while(row = await rs.next()) {
    console.log("here's a row:", row);
}

我以为我可以使用生成器来控制它，但看起来你不能在回调中使用 yield。如果你仔细想想，这实际上似乎是合乎逻辑的。

class ApiWrapper {
    constructor(api) {
        this.api = api;
    }
    prepare() {
        return new Promise((resolve, reject) => {
            this.api.prepare((err, rs) => {
                if (err) {
                    reject(err);
                } else {
                    resolve(rs);
                }
            });
        });
    }
    *rows() {
        this.api.fetchRows((err, row) => {
            if (err) {
                throw err;
            } else {
                yield row; // nope, not allowed here
            }
        });
    }
    next() { ... }
}

那么我有什么选择呢？

重要提示：我不想在数组中存储任何内容然后对其进行迭代，我们在这里讨论的是千兆负载的行数据。

编辑 我可以使用stream.Readable 模拟我想要的行为，但它警告我这是一个实验性功能。这是我尝试使用stream 解决的基于数组的简化版本：

const stream = require('stream');
function gen(){
    const s = new stream.Readable({
        objectMode: true,
        read(){
            [11, 22, 33].forEach(row => {
                this.push({ value: row });
            });
            this.push(null)
        }
    });
    return s;
}

for await (let row of gen()) {
    console.log(row);
}

// { value: 11 }
// { value: 22 }
// { value: 33 }

(node:97157) ExperimentalWarning: Readable[Symbol.asyncIterator] is an experimental feature. This feature could change at any time

Answer 1

我终于意识到我需要类似于 Go 的频道，并且与 async/await 兼容。基本上答案是同步一个异步迭代器和一个回调，让它们在 next() 迭代被消耗时相互等待。

我发现的最佳（Node）native 解决方案是使用 stream 作为迭代器，它在 Node 10.x 中受支持，但标记为实验性的。我还尝试使用p-defer NPM 模块来实现它，但结果比我预期的要复杂。终于碰到了https://www.npmjs.com/package/@nodeguy/channel 模块，这正是我所需要的：

const Channel = require('@nodeguy/channel');

class ApiWrapper {
    // ...
    rows() {
        const channel = new Channel();
        const iter = {
            [Symbol.asyncIterator]() {
                return this;
            },
            async next() {
                const val = await channel.shift();
                if (val === undefined) {
                    return { done: true };
                } else {
                    return { done: false, value: val };
                }
            }
        };

        this.api.fetchRows(async (err, row) => {
            await channel.push(row);
        }).then(() => channel.close());

        return iter;
    }
}

// then later

for await (let row of rs.rows()) {
     console.log(row)
}

注意每个迭代函数核心next() 和rows() 有一个await 来限制可以通过通道推送的数据量，否则生产回调最终可能会无法控制地将数据推送到通道队列中.这个想法是回调应该等待数据被迭代器next()消费，然后再推送更多。

这是一个更独立的例子：

const Channel = require('@nodeguy/channel');

function iterating() {
    const channel = Channel();

    const iter = {
        [Symbol.asyncIterator]() {
            return this;
        },
        async next() {
            console.log('next');
            const val = await channel.shift();
            if (val === undefined) {
                return { done: true };
            } else {
                return { done: false, value: val };
            }
        }
    };

    [11, 22, 33].forEach(async it => {
        await channel.push(it);
        console.log('pushed', it);
    });

    console.log('returned');
    return iter;
}

(async function main() {
    for await (let it of iterating()) {
        console.log('got', it);
    }
})();

/*
returned
next
pushed 11
got 11
next
pushed 22
got 22
next
pushed 33
got 33
next
*/

就像我说的，可以使用 Streams 和/或 Promises 来实现这一点，但 Channel 模块解决了一些复杂性，使其更直观。