javascript - 如何在不泄漏 promise 的情况下将流转换为生成器

Question

我有一个流，我需要将其转换为生成器，以便上传者可以使用通用生成器。

这意味着转向:

stream.on('data', chunk => ...);

到:

generator = streamGenerator(stream);
chunk = await generator.next()
...

更好的是:

chunk = yield streamGenerator;

总的来说，我最好的尝试需要从 promise 中泄露决心，我想避免这种情况:

function streamToIterable(chunkSize, stream) {
    let collector = [];
    let value = [];
    let done = false;
    let _resolve;
    let promise = new Promise(resolve => _resolve = resolve);
    stream.on('data', chunk => {
        collector = collector.concat(chunk);
        if (value.length >= chunkSize) {
            value = collector.splice(0, chunkSize);
            _resolve(value);
            stream.pause();
        }
    });
    stream.on('end', () => {
        _resolve(collection);

        // With done set to true, the next iteration well ignore 'value' and end the loop
        done = true;
    });
    stream.resume();

    return {
        next: () => ({
            value: promise.then(() => {
                stream.resume();
                promise = new Promise(resolve => _resolve = resolve);
            }),
            done,
        }),
    };
}

function* streamToGenerator(stream) {
    const iterator = streamToIterable(stream);
    let next = iterator.next();
    while (!next.done) {
        yield next.value;
    }
};

在生成器中用于上传 block 的用法:

for (const chunkData of generator()) {
    let result = yield uploadPost(url, formData, onChunkProgress(chunkIndex));

这是在 redux-saga 中，所以在 return promise 被 resolved 之前不会在生成器上调用“next()”。

Answer 1

如果您想使用单个事件监听器来解决不同的 promise ，则无法避免将 resolve 函数存储在可变变量中。您可以使用 once method 简化 promise 创建类似于以下内容:

function streamToIterator(stream) {
    let done = false;
    const end = new Promise(resolve => {
        stream.once('end', resolve);
    }).then(e => {
        done = true;
    });

    return {
        [Symbol.iterator]() { return this; }
        next() {
            const promise = new Promise(resolve => {
                stream.once('data', value => {
                    resolve(value);
                    stream.pause();
                });
                stream.resume();
            });

            return {
                value: Promise.race([promise, end]),
                done,
            };
        }),
    };
}

当然，您正在自己进行 end 和 data 之间的竞赛，您在第一次调用 next 之前恢复流，并且最重要的是，您自己进行分块，因此这可能适用于您的情况。

除此之外，我建议查看 buffering internals of node.js streams ，使用较低级别的 API 可能比 data 事件更容易读取特定大小的 block 。

另外你绝对应该看看 asynchronous iteration proposal对于 es-next。您尝试实现的可迭代接口(interface)非常相似，而且他们肯定已经拥有或真的会欢迎一个使节点可读流可迭代的示例。