Javascript - 分叉异步生成器[重复]答案

【问题标题】：Javascript - Fork async generator [duplicate]Javascript - 分叉异步生成器[重复]
【发布时间】：2021-11-12 19:31:36
【问题描述】：

假设我有一个异步生成器，如下所示：

// This could be records from an expensive db call, for example...
// Too big to buffer in memory
const events = (async function* () {
    await new Promise(r => setTimeout(r, 0));
    yield {type:'bar', ts:'2021-01-01 00:00:00', data:{bar:"bob"}};
    yield {type:'foo', ts:'2021-01-02 00:00:00', data:{num:2}};
    yield {type:'foo', ts:'2021-01-03 00:00:00', data:{num:3}};
})();

我怎样才能复制它来达到类似的效果：

function process(events) {

    async function* filterEventsByName(events, name) {
        for await (const event of events) {
            if (event.type === name) continue;
            yield event;
        }
    }

    async function* processFooEvent(events) {
        for await (const event of events) {
            yield event.data.num;
        }
    }

    // How to implement this fork function?
    const [copy1, copy2] = fork(events);

    const foos = processFooEvent(filterEventsByName(copy1, 'foo'));
    const bars = filterEventsByName(copy2, 'bar');

    return {foos, bars};
}

const {foos, bars} = process(events);

for await (const event of foos) console.log(event);
// 2
// 3

for await (const event of bars) console.log(event);
// {type:'bar', ts:'2021-01-01 00:00:00', data:{bar:"bob"}};

【问题讨论】：

作为一个生成器，只实例化一次是没有意义的。只需保持生成器函数不变，并将其传递到filterByName。
@MarioVernari 为了这个例子，我只是给出了一个简单的生成，实际上生成器是从一个我只想进行一次的昂贵的 API 调用创建的。
那么也许只有实际的 API 调用应该进行一次并缓存，而生成器应该仍然被实例化两次？
@CherryDT API 结果集太大而无法放入内存。
但这不合逻辑。让我们假设有一种复制生成器状态的方法（没有）。那这意味着什么？如果都是一个请求，那么响应将存储在哪里，如果不在内存中？如果以分页方式根据需要完成了多个请求，那么如果其中一个生成器副本已经被 next 调用了 100 次而另一个被调用了 500 次，会发生什么？考虑到这一点，您会得出结论，唯一有意义的就是必须分离生成器。

标签： javascript generator

【解决方案1】：

我有一个使用Highland 作为中介的解决方案。

请注意（来自文档）：

一个分叉到多个消费者的流将一次一个地从其源中提取值，只要最慢的消费者可以处理它们。

import _ from 'lodash'
import H from 'highland'

export function fork<T>(generator: AsyncGenerator<T>): [
    AsyncGenerator<T>,
    AsyncGenerator<T>
] {
    const source = asyncGeneratorToHighlandStream(generator).map(x => _.cloneDeep(x));
    return [
        highlandStreamToAsyncGenerator<T>(source.fork()),
        highlandStreamToAsyncGenerator<T>(source.fork()),
    ];
}

async function* highlandStreamToAsyncGenerator<T>(
    stream: Highland.Stream<T>
): AsyncGenerator<T> {
    for await (const row of stream.toNodeStream({ objectMode: true })) {
        yield row as unknown as T;
    }
}

function asyncGeneratorToHighlandStream<T>(
    generator: AsyncGenerator<T>
): Highland.Stream<T> {
    return H(async (push, next) => {
        try {
            const result = await generator.next();
            if (result.done) return push(null, H.nil);
            push(null, result.value);
            next();
        } catch (error) {
            return push(error);
        }
    });
}

希望看到没有库或其他库的替代解决方案。

【讨论】：