Typescript - Async Queue Example

README.md

Typescript - Async Queue Example

This example demonstrates how to use the queue function from the async package in a Node.js project with TypeScript. The code sets up a monitor to visualize the behavior of the queue and executes a "worker" function whenever the queue processes an item.

Why

The queue function from the async library is useful when your application has many tasks to handle, but does not require a persistent queue. This can be particularly beneficial for managing asynchronous operations such as API requests or database queries in a controlled manner, ensuring that tasks are processed concurrently without overwhelming system resources.

Features

1. Utilizes the queue function from the async package with Promises.
2. Monitors and visualizes the queue's behavior.
3. Processes various asynchronous tasks through a customizable "worker" function.

Installation

To install the necessary library in a TypeScript project, you can use one of the following package managers:

npm

npm install async

pnpm

pnpm add async

yarn

yarn add async

Running the Code

You can run the code using any TypeScript transpiler such as ts-node or swc-node. Here are examples for each:

Using ts-node

npx ts-node your-script.ts
yarn ts-node your-script.ts
pnpm ts-node your-script.ts

Using swc-node

node -r @swc-node/register your-script.ts

Replace your-script.ts with the path to your TypeScript file.

example.ts

import async from 'async';

const ONE_SECOND_IN_MS = 1e3;
const lang = 'pt-BR',
  maxConcurrentTasks = 2,
  maxProcessingTimeInMs = ONE_SECOND_IN_MS * 5,
  minProcessingTimeInMs = 200,
  monitorRefreshTimeInMs = minProcessingTimeInMs,
  maxHistoryInScreen = 10,
  history: string[] = [],
  maxTasksToProcess = 100,
  separator = Array(54).fill('_').join('');
let completedTasksCount = 0;

const interval = setInterval(() => {
  console.clear();
  console.log(
    `${queue.length()} tasks waiting | ${queue.running()} running tasks | ${completedTasksCount} completed tasks`,
  );
  console.log(
    `Time for processing | Min: ${maskTime(minProcessingTimeInMs)} | Max: ${maskTime(maxProcessingTimeInMs)}`,
  );
  console.log(separator + '\n');

  const historyStart = Math.max(history.length - maxHistoryInScreen, 0);

  history.slice(historyStart).forEach((h) => console.log(h));
  if (queue.length() === 0 && completedTasksCount === maxTasksToProcess) {
    console.log('All tasks processed');
    clearInterval(interval);
  }
}, monitorRefreshTimeInMs);

function maskTime(time: number) {
  return time > ONE_SECOND_IN_MS ? `${time / ONE_SECOND_IN_MS}s` : `${time}ms`;
}

function getCurrentTime() {
  const currentDate = new Date();
  const currentTime = currentDate.toLocaleTimeString(lang);

  return {
    currentDate,
    currentTime,
  };
}

function makePrinter(number: number) {
  let startDate: Date;
  return {
    printStart() {
      const { currentTime, currentDate } = getCurrentTime();
      startDate = currentDate;
      history.push(`${currentTime}: Task started: #${number}`);
    },

    printEnd() {
      const { currentDate, currentTime } = getCurrentTime();
      const diff = currentDate.getTime() - startDate.getTime();
      const diffMask = maskTime(diff);
      history.push(
        `${currentTime}: Finished task #${number} after ${diffMask} of processing`,
      );
    },
  };
}

function pause() {
  return new Promise((resolve) =>
    setTimeout(
      resolve,
      Math.random() * maxProcessingTimeInMs + minProcessingTimeInMs,
    ),
  );
}

async function worker(number: number) {
  const { printStart, printEnd } = makePrinter(number);
  printStart();
  await pause();
  printEnd();
  completedTasksCount++;
}

const queue = async.queue(
  worker, // Async function, can be HTTP request, database query, etc
  maxConcurrentTasks,
);

Array.from({ length: maxTasksToProcess }).forEach((_, i) => {
  queue.push(i);
});