Convert function using callbacks into Async Iterator variant

This has been a NodeJS-native API since v10, no need reinventing it:

const {createReadStream} = require('fs');
const {createInterface} = require('readline');

function readFileLines(fileName: string): AsyncIterable<string> {
    const input = createReadStream(fileName);
    return createInterface({input, crlfDelay: Infinity});
}

Testing it:

const lines = readFileLines('./test1.js');
for await(const l of lines) {
    console.log(l);
}

Yes.

I did this for Deno.serve which is an HTTP server that takes a callback and an options object like Deno.serve(req => respondWith(req), {port: 3000}).

Basically the code is;

async function* emitterGen(opts){
  let _resolve,
      _req = new Promise((resolve,reject) => _resolve = resolve);
  Deno.serve( req => ( _resolve(req)
                     , _req = new Promise((resolve,reject) => _resolve = resolve)
                     )
            , opts
            );
  while (true){
    yield await _req;
  }
}

let reqEmitter = emitterGen({port: 3000});

for await (let req of reqEmitter){
  respondWith(req);
}

Obviously the code above is simplified, without exception handling. Yet it should be sufficient to answer your question.

Here is a working mock up sever which creates a random number (0-99) as request (req) at every random (0-999) ms and invokes cb (handler) with req. Stops after 5 iterations.

function server(cb,ms){
  let count  = 5,
      looper = function(c = count,t = ms){
                 let stoid = setTimeout( req => ( cb(req)
                                                , --c && looper(c, Math.random()*1000 >>> 0)
                                                , clearTimeout(stoid)
                                                )
                                       , t
                                       , Math.random()*100 >>> 0
                                       )
               }
  looper();
}

async function* emitterGen(ms){
  let _resolve,
      _req = new Promise((resolve,reject) => _resolve = resolve);
  server( req => ( _resolve(req)
                 , _req = new Promise((resolve,reject) => _resolve = resolve)
                 )
        , ms
        );
  while (true){
    yield await _req;
  }
}

let reqEmitter = emitterGen(1000);

// since there is no top level await in SO snippets
(async function(){
  for await (let req of reqEmitter){
    console.log(`Received request is: ${req}`);
  }
})();

I created a class that can produce an async generator from any source:

/** Infinite async generator. Iterates messages pushed to it until closed. */
class Machina<T> {

  #open = true;
  #queue: T[] = [];
  #resolve: (() => void) | undefined;

  async * stream(): AsyncGenerator<T> {
    this.#open = true;

    while (this.#open) {
      if (this.#queue.length) {
        yield this.#queue.shift()!;
        continue;
      }

      await new Promise<void>((_resolve) => {
        this.#resolve = _resolve;
      });
    }
  }

  push(data: T): void {
    this.#queue.push(data);
    this.#resolve?.();
  }

  close(): void {
    this.#open = false;
    this.#resolve?.();
  }

}

export { Machina };

You can use it like this:

// Create the Machina instance
const machina = new Machina<string>();

// Async generator loop
async function getMessages() {
  for await (const msg of machina.stream()) {
    console.log(msg);
  }
}

// Start the generator
getMessages();

// Push messages to it
machina.push('hello!');
machina.push('whats up?');
machina.push('greetings');

// Stop the generator
machina.close();

For your specific case, something like this should work:

/** Read each line of the file as an AsyncGenerator. */
function readFileAsync(path: string): AsyncGenerator<string> {
  const machina = new Machina<string>();

  readFile(path, (line: string, eof: boolean) => {
    if (eof) {
      machina.close();
    } else {
      machina.push(line);
    }
  });

  return machina.stream();
}

// Usage
for await (const line of readFileAsync('file.txt')) {
  console.log(line);
}

How it works

1. Calling machina.stream() kicks off an infinite loop, but it becomes paused immediately (on the first iteration) because it's waiting for a promise that isn't resolved.
2. Calling machina.push() adds an item to the buffer, and then unpauses it by resolving the promise. When it becomes unpaused, it empties the buffer into the stream and then pauses it again by awaiting a new promise.
3. The consumer of machina.stream() receives the pushed items. You can do this repeatedly.

Other considerations:

• Even if the Machina instance goes out of scope (ie ready to be garbage-collected), its promise will still run forever. So you need to manually call machina.close() when you are done streaming, if ever. Simply breaking out of the loop is not enough!
• I tried a slightly different design at first, without using a buffer. It turns out the buffer is needed, otherwise you cannot push multiple items in the same tick (all after the first will be dropped). But if you're only pushing one item per tick and are actively consuming the stream, the buffer will only ever contain one item. This is basically just a small memory consideration.