Processing tasks as they are completed

I was reading some Python code and really like the signature of asyncio.as_completed(). It returns an iterable of tasks that you can await. What’s really nice about this is that it’s very intuitive. The first task that you await is the first task that completes.

I’ve decided to implement a version of it in typescript.

Setup

We want a way to generate a list of async tasks with random delays. The following should do that

function genRandomDelay() {
  return Math.floor(Math.random() * 1000);
}

function genAsyncTask(): Promise<number> {
  const delay = genRandomDelay();
  return new Promise((resolve) => {
    setTimeout(() => {
      resolve(delay);
    }, delay);
  });
}

const tasks = Array.from({ length: 10 }, genAsyncTask);

Now the goal is to process these tasks as they are completed. Here process just means printing out their result. So we should they the numbers are printed out in ascending order.

Implementation

At a high level, our function asCompleted should return a list of promisses. When a task is completed, it should resolve the next promise in the list. So we need to keep a next state that we could increment each time a task is completed.

function asCompleted<T>(tasks: Promise<T>[]): Promise<T>[]{
  let next = 0;
  const promises = Array.from({length: tasks.length}, withResolvers<T>);
  tasks.forEach((task) => {
    task.then((result) => {
      const {resolve} = promises[next];
      next++;
      resolve(result);
    }).catch(err => {
      const {reject} = promises[next];
      next++;
      reject(err);
    })
  });

  return promises.map(({promise}) => promise);
}

Here withResolveers() is a function that returns {promise, resolve, reject}. It’s basically Promise.withResolvers() in ES2024, but the polyfill looks something like this

function withResolvers<T>() {
  let resolve!: (value: T | PromiseLike<T>) => void;
  let reject!: (reason?: any) => void;
  const promise = new Promise<T>((res, rej) => {
    resolve = res;
    reject = rej;
  });
  return { promise, resolve, reject };
}

Usage

This is a sample usage

async function main() {
  const tasks = Array.from({ length: 10 }, genAsyncTask);
  for (const promise of asCompleted(tasks)) {
    const result = await promise;
    console.log(result);
  }
}

main().catch((error) => {
  console.error('Error:', error);
});