Transform a callback-based function into an Async Iterator version

Question

Transform a callback-based function into an Async Iterator version

Situation

I have a function with an asynchronous callback structure, like so:

let readFile:   (path: string, callback: (line: string, eof: boolean) => void) => void

However, I would much rather use a function that follows the AsyncIterable/AsyncGenerator pattern:

let readFileV2: (path: string) => AsyncIterable<string>

Issue

Without readFileV2, I'm stuck reading a file in this cumbersome manner:

let file = await new Promise((res, err) => {
    let file = ''
    readFile('./myfile.txt', (line, eof) => {
        if (eof) { return res(file) }
        file += line + '\n'
    })
})

.. whereas with readFileV2, I can elegantly achieve the same result like so:

let file = '';
for await (let line of readFileV2('./myfile.txt')) {
    file += line + '\n'
}

Inquiry

~~Is there a way for me to transform readFile into readFileV2?~~

Updated for clarity:

Is there a method that can be applied universally to convert a function with an async callback into an AsyncGenerator/AsyncIterable format?

If so, could this method be demonstrated using the readFile function mentioned above?

References

I have come across a couple of related questions:

How to convert Node.js async streaming callback into an async generator?
How to convert callback-based async function to async generator

However, none of these seem to offer a definitive solution.

javascript typescript asynccallback async-iterator

Answer 1

Answer №1

Let me start with a disclaimer: I will be addressing the following question:

How can we modify a data-providing function fn that takes arguments of type A and a callback function
(data: T, done: boolean) => void
to transform it into a new function transform(fn) with a signature of
(...args: A) => AsyncIterable<T>
?

While this transformation might not be ideal in all scenarios, as the processing of AsyncIterable<T> data could stall or terminate prematurely, the implementation provided below showcases one way this can be achieved.

Here is the proposed implementation:

// Code implementation goes here

This implementation effectively manages a queue of data values, values, that are manipulated within the callback function of fn and consumed by a generator function. The process involves a series of promises to handle the flow of data from fn to the generator function.

To test this implementation, an initial fn needs to be provided. A sample function provideData is demonstrated below:

// Sample data-providing function goes here

The function provideData essentially delivers lines of data to a callback function at a one-second interval. This function is then passed through the transform function to create provideDataV2, an AsyncIterable data provider.

// Transformation of provideData to provideDataV2

The testing of this transformation is done by executing a function that iterates over the data provided by provideDataV2 and logs each line with a timestamp. The log output confirms the successful transformation.

While this implementation serves the purpose outlined in the initial question, it may not cover all edge cases or error handling scenarios. It's advisable to explore other recommended solutions and evaluate the functionality based on specific requirements.

Explore the code on TypeScript Playground

Answer 2

Let me start with a disclaimer: I will be addressing the following question:

How can we modify a data-providing function fn that takes arguments of type A and a callback function
(data: T, done: boolean) => void
to transform it into a new function transform(fn) with a signature of
(...args: A) => AsyncIterable<T>
?

While this transformation might not be ideal in all scenarios, as the processing of AsyncIterable<T> data could stall or terminate prematurely, the implementation provided below showcases one way this can be achieved.

Here is the proposed implementation:

// Code implementation goes here

This implementation effectively manages a queue of data values, values, that are manipulated within the callback function of fn and consumed by a generator function. The process involves a series of promises to handle the flow of data from fn to the generator function.

To test this implementation, an initial fn needs to be provided. A sample function provideData is demonstrated below:

// Sample data-providing function goes here

The function provideData essentially delivers lines of data to a callback function at a one-second interval. This function is then passed through the transform function to create provideDataV2, an AsyncIterable data provider.

// Transformation of provideData to provideDataV2

The testing of this transformation is done by executing a function that iterates over the data provided by provideDataV2 and logs each line with a timestamp. The log output confirms the successful transformation.

While this implementation serves the purpose outlined in the initial question, it may not cover all edge cases or error handling scenarios. It's advisable to explore other recommended solutions and evaluate the functionality based on specific requirements.

Explore the code on TypeScript Playground

Answer 3

Answer №2

Ever since version 10, this NodeJS-native API has been available, eliminating the need to recreate it:

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

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

Let's give it a try:

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

Answer 4

Ever since version 10, this NodeJS-native API has been available, eliminating the need to recreate it:

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

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

Let's give it a try:

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

Answer 5

Answer №3

Affirmative.

I utilized this approach with Deno.serve, an HTTP server that accepts a callback and an options object in the form of

Deno.serve(req => respondWith(req), {port: 3000})

.

The gist of the code is as follows;

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);
}

The code provided above is a simplified version, lacking exception handling, but it should suffice to address your query.

Presented here is a functional simulated server that generates a random number (0-99) as a request (req) every random (0-999) ms, triggering the invocation of cb (handler) with req. It halts 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);

// due to the absence of top-level await in SO snippets
(async function(){
  for await (let req of reqEmitter){
    console.log(`Received request is: ${req}`);
  }
})();

Answer 6

Affirmative.

I utilized this approach with Deno.serve, an HTTP server that accepts a callback and an options object in the form of

Deno.serve(req => respondWith(req), {port: 3000})

.

The gist of the code is as follows;

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);
}

The code provided above is a simplified version, lacking exception handling, but it should suffice to address your query.

Presented here is a functional simulated server that generates a random number (0-99) as a request (req) every random (0-999) ms, triggering the invocation of cb (handler) with req. It halts 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);

// due to the absence of top-level await in SO snippets
(async function(){
  for await (let req of reqEmitter){
    console.log(`Received request is: ${req}`);
  }
})();

Answer 7

Answer №4

I have developed a unique class that has the ability to create an asynchronous generator from any given source:

/** Implementation of an endless async generator. Processes incoming messages until terminated. */
class Automaton<T> {

  #active = true;
  #buffer: T[] = [];
  #resolve: (() => void) | undefined;

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

    while (this.#active) {
      if (this.#buffer.length) {
        yield this.#buffer.shift()!;
        continue;
      }

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

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

  end(): void {
    this.#active = false;
    this.#resolve?.();
  }

}

export { Automaton };

You can utilize it in the following manner:

// Instantiate the Automaton
const automaton = new Automaton<string>();

// Asynchronous generator loop
async function processMessages() {
  for await (const message of automaton.generate()) {
    console.log(message);
  }
}

// Initialize the generator
processMessages();

// Add messages to the generator
automaton.add('hello!');
automaton.add('how are you?');
automaton.add('salutations');

// Terminate the generator
automaton.end();

For your specific scenario, consider integrating it like this:

/** Read each line of a file as an AsyncGenerator. */
function readFileSync(path: string): AsyncGenerator<string> {
  const automaton = new Automaton<string>();

  readFile(path, (line: string, eof: boolean) => {
    if (eof) {
      automaton.end();
    } else {
      automaton.add(line);
    }
  });

  return automaton.generate();
}

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

Functionality Overview

Invoking automaton.generate() triggers an infinite loop, although it pauses immediately (on the first iteration) while awaiting a promise resolution.
Utilizing automaton.add() appends an item to the buffer and then resumes by resolving the promise. Upon resuming, it transfers the buffer contents to the stream before pausing again via a new promise.
The consumer of automaton.generate() receives the added items, and this process can be repeated.

Additional Remarks:

Even if the Automaton instance is no longer in use (potentially subject to garbage collection), its promise continues indefinitely. Therefore, remember to call automaton.end() manually once streaming is complete. Simply exiting the loop is insufficient!
Initially, I experimented with an alternative design sans a buffer. However, it became apparent that a buffer is necessary to accommodate the addition of multiple items within a single tick; otherwise, subsequent items beyond the first would be lost. Nonetheless, if only one item is added per tick and the stream is actively consumed, the buffer will maintain a single item at most—a minor memory consideration.

Answer 8

I have developed a unique class that has the ability to create an asynchronous generator from any given source:

/** Implementation of an endless async generator. Processes incoming messages until terminated. */
class Automaton<T> {

  #active = true;
  #buffer: T[] = [];
  #resolve: (() => void) | undefined;

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

    while (this.#active) {
      if (this.#buffer.length) {
        yield this.#buffer.shift()!;
        continue;
      }

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

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

  end(): void {
    this.#active = false;
    this.#resolve?.();
  }

}

export { Automaton };

You can utilize it in the following manner:

// Instantiate the Automaton
const automaton = new Automaton<string>();

// Asynchronous generator loop
async function processMessages() {
  for await (const message of automaton.generate()) {
    console.log(message);
  }
}

// Initialize the generator
processMessages();

// Add messages to the generator
automaton.add('hello!');
automaton.add('how are you?');
automaton.add('salutations');

// Terminate the generator
automaton.end();

For your specific scenario, consider integrating it like this:

/** Read each line of a file as an AsyncGenerator. */
function readFileSync(path: string): AsyncGenerator<string> {
  const automaton = new Automaton<string>();

  readFile(path, (line: string, eof: boolean) => {
    if (eof) {
      automaton.end();
    } else {
      automaton.add(line);
    }
  });

  return automaton.generate();
}

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

Functionality Overview

Invoking automaton.generate() triggers an infinite loop, although it pauses immediately (on the first iteration) while awaiting a promise resolution.
Utilizing automaton.add() appends an item to the buffer and then resumes by resolving the promise. Upon resuming, it transfers the buffer contents to the stream before pausing again via a new promise.
The consumer of automaton.generate() receives the added items, and this process can be repeated.

Additional Remarks:

Even if the Automaton instance is no longer in use (potentially subject to garbage collection), its promise continues indefinitely. Therefore, remember to call automaton.end() manually once streaming is complete. Simply exiting the loop is insufficient!
Initially, I experimented with an alternative design sans a buffer. However, it became apparent that a buffer is necessary to accommodate the addition of multiple items within a single tick; otherwise, subsequent items beyond the first would be lost. Nonetheless, if only one item is added per tick and the stream is actively consumed, the buffer will maintain a single item at most—a minor memory consideration.

Transform a callback-based function into an Async Iterator version

Situation

Issue

Inquiry

References

Answer №1

Answer №2

Answer №3

Answer №4

Functionality Overview

Similar questions

What is the process for implementing a third-party component in my web application?

The function 'downloadFunc' is not recognized as a valid function within the context of ReactJS custom hooks

Angular version 5 and above introduces a new feature called "openFromComponent" within the Snackbar component, facilitating seamless communication

Looking for assistance with transferring a data attribute to a form redirection

In order to verify the dynamic components within the table

What causes an undefined outcome when a promise is fulfilled?

The act of employing `Function.prototype.run` within an Angular TypeScript class is deemed as illegitimate

I'd really appreciate your assistance in obtaining a value for a specific webelement

Installing material-ui using npm does not always result in getting the most up-to-date version

Error message: Upon refreshing the page, the React Router is unable to read properties of

What is the best way to divide multiple event handlers in jQuery?

Adding functions to the window scroll event

Checking if the group of radio buttons has been selected using JQUERY

Is there a more efficient algorithm available to solve this problem in a quicker manner?

Issue with arrow functions in Reactjs Material-UI - Unexpected token error

Performing mathematical operations in JavaScript, rounding to the nearest .05 increment with precision up to two

The URL may change, but the component remains constant when navigating back

Creating a responsive HTML, CSS, and JavaScript popup: A step-by-step guide

Issue with Component: Data is not being injected into controller from ui-router resolve, resulting in undefined data

Transmit both image and JSON data in a single response using Node.js and Express