Say I have an async generator, like this:
// 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}};
})();
How can I copy it to acheive something like:
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"}};
I have a solution using Highland as an intermediary.
Note that (from the docs):
A stream forked to multiple consumers will pull values, one at a time, from its source as only fast as the slowest consumer can handle them.
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);
}
});
}
Would love to see alternative solutions without a library, or with another library.