如何将流转换为生成器而不会泄漏诺言的解决方案 [英] How to convert a stream into a generator without leaking resolve from a promise

问题描述

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

I have a stream and I need to convert it to a generator, so an uploader can consume the generic generator.

这意味着转向:

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

收件人:

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

更好:

chunk = yield streamGenerator;

总的来说，我最好的尝试是从诺言中泄漏决心，我想避免这种情况:

Overall my best attempt requires leaking the resolve from a promise and I'd like to avoid that:

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;
    }
};

在生成器中使用以上传块:

Usage in a generator for uploading chunks:

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

这是在redux-saga中的，因此在解决了返回承诺之前，不会在生成器上调用"next()".

This is in a redux-saga, so "next()" isn't called on the generator until the return promise is resolved.

推荐答案

如果要使用可解析不同承诺的单个事件侦听器，则无法避免将resolve函数存储在可变变量中.您可以使用 once方法来简化承诺的创建与以下内容类似:

You cannot avoid storing the resolve function in a mutable variable if you want to use a single event listener that resolves different promises. You could simplify the promise creation by using the once method similar to the following:

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之前恢复流，最重要的是您自己进行分块，因此这可能适用于您的情况.

Of course, you are doing the racing between end and data yourself, you resume the stream before next is called the first time and most importantly you do the chunking yourself, so this might to be applicable to your situation.

除此之外，我建议您查看缓冲node.js流的内部信息，与data事件相比，使用较低级别的API读取某些大小的块可能会更容易.

Apart from that, I'd recommend to check out the buffering internals of node.js streams, it might be easier to read chunks of certain sizes using a lower-level API than data events.

此外，您绝对应该查看es-的异步迭代提案.下一个.您尝试实现的可迭代接口非常相似，并且可以肯定的是，它们已经或确实会欢迎使节点可读流可迭代的示例.

Also you definitely should have a look at the asynchronous iteration proposal for es-next. The iterable interface you're trying to implement is very similar, and surely they either already have or really would welcome an example of making a node readablestream iterable.

这篇关于如何将流转换为生成器而不会泄漏诺言的解决方案的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！