AsyncIterator

main.ts

type Emit<T> = (value: T) => void;
type Subscribe<T> = (emit: Emit<T>) => void | (() => void);

export async function* toAsyncIterator<T>(subscribe: Subscribe<T>): AsyncGenerator<T> {
    const queue = new AsyncQueue<T>();
    const cleanup = subscribe(v => queue.push(v));

    try {
        while (true) {
            yield queue.next();
        }
    } finally {
        cleanup?.();
    }
}

export class AsyncQueue<T> {
    private buffer: Array<{ value: T }> = [];
    private pending: Array<(item: { value: T }) => void> = [];

    push(value: T) {
        const next = this.pending.shift();
        if (next) return next({ value });

        this.buffer.push({ value });
    }

    async next(): Promise<T> {
        const item = this.buffer.shift();
        if (item) return item.value;

        return new Promise<T>(resolve => {
            this.pending.push(({ value }) => resolve(value));
        });
    }
}

async function main() {
    const interval = setInterval(() => console.log('tick', Date.now()), 100);

    const iter = toAsyncIterator<number>(emit => {
        let i = 0;
        const id = setInterval(() => emit(++i), 500);
        return () => clearInterval(id);
    });

    for await (const value of iter) {
        console.log('yielded:', value);
        if (value >= 3) break;
    }

    clearInterval(interval);
    console.log('done');
}

main();