I have a function that streams data in batches via a callback.

Each batch will await the callback function before fetching another batch and the entire function returns a promise that resolves when all batches are finished.

(I'm using TypeScript annotations to help with readability)

async function callbackStream(fn: (batch: Array) => Promise) {}

How do I to turn this function into an async generator that yields one value at a time?

async function* generatorStream(): AsyncIterableIterator {}

This has proven to be quite a difficult task.

I've toyed around with this problem and I've built something that works, but its very convoluted and I can't justify merging this code and making others on my team deal with it.

Here's my current implementation:

I'm using this helper function that created a "deferred" promise which helps with passing promises around callbacks.

interface DeferredPromise {
    resolve: (value: T) => void
    reject: (error: any) => void
    promise: Promise
}

function deferred(): DeferredPromise {
    let resolve
    let reject
    const promise = new Promise((res, rej) => {
        resolve = res
        reject = rej
    })
    return {
        resolve: resolve as (value: T) => void,
        reject: reject as (error: any) => void,
        promise,
    }
}

Next I have this hairball of logic that linearizes the promise callbacks into a chain where each promise resolved a batch with next function that will return another promise fetching the next batch.

type Done = { done: true }
type More = { done: false; value: Array; next: () => Promise }
type Result = More | Done

async function chainedPromises() {
    let deferred = PromiseUtils.deferred()

    callbackStream(async batch => {
        const next = PromiseUtils.deferred()
        deferred.resolve({
            done: false,
            value: batch,
            next: () => {
                deferred = PromiseUtils.deferred()
                next.resolve(null)
                return deferred.promise
            },
        })
        await next.promise
    }).then(() => {
        deferred.resolve({ done: true })
    })

    return deferred.promise
}

From here, creating a generator that yields one item at a time isn't very difficult:

async function* generatorStream(): AsyncIterableIterator {
    let next = chainedPromises
    while (true) {
        const result = await next()
        if (result.done) {
            return
        }
        for (const item of result.value) {
            yield item
        }
        next = result.next
    }
}

I think we can all agree that the intermediate chainedPromises function is very confusing and convoluted. Is there any way I can transform callbackStream into generatorStream in a way that is easy to understand and easy to follow? I don't mind using a library if its well established, but I would also appreciate a simple implementation from first-principles.