createAsyncGenerator.ts

export async function* createAsyncGenerator<T>(
    createResource: (opts: { send: (value: T) => void, close: () => void }) => void,
    reconnect?: () => Promise<void>
): AsyncGenerator<T, void, undefined> {
    while (true) {
        let closed = false;
        let buffer: T[] = [];
        let bufferHasData = createEvent();

        createResource({
            send(data) {
                buffer.push(data);
                bufferHasData.set();
                bufferHasData = createEvent();
            },
            close() {
                closed = true;
                bufferHasData.set();
            }
        });

        while (!closed) {
            yield* buffer;
            await bufferHasData.wait;
        }

        if (typeof reconnect !== "undefined") {
            await reconnect();
        }
    }
}

export function createEvent() {
    let resolve: ((value?: void) => void) | true | null = null;
    return {
        set() {
            if (resolve === null || resolve === true) {
                resolve = true;
            } else {
                resolve();
            }
        },
        wait: new Promise<void>(r => {
            if (resolve === true) {
                r();
            } else {
                resolve = r;
            }
        })
    };
}