{"id":31378,"date":"2025-02-10T11:08:09","date_gmt":"2025-02-10T10:08:09","guid":{"rendered":"https:\/\/www.codemotion.com\/magazine\/?p=31378"},"modified":"2025-02-10T11:12:33","modified_gmt":"2025-02-10T10:12:33","slug":"queueing-without-a-queue-enter-fastq","status":"publish","type":"post","link":"https:\/\/www.codemotion.com\/magazine\/backend\/software-architecture\/queueing-without-a-queue-enter-fastq\/","title":{"rendered":"Queueing Without a Queue: Enter fastq"},"content":{"rendered":"\n

How often have you needed to decouple processes using a queue, only to find that you didn’t have one due to limitations such as resource constraints<\/strong> or the inability to install additional software?<\/p>\n\n\n\n

In this series, I’ll show you how to solve this problem by using different libraries of tools that help you resolve it even if you do not have a real queue. These solutions are versatile and can be applied to a variety of scenarios.<\/p>\n\n\n\n

Before getting started, I want to reassure you: these approaches are not just for when you cannot have a real queue due to limitations. They are also flexible enough<\/strong> to be used even when you can have one, without any problem.
By exploring the examples in this series, you’ll evaluate which approach is the best for your case.<\/p>\n\n\n\n

Fastq<\/h2>\n\n\n\n

The first post is about fastq<\/a>, an npm package for Node.js<\/a> built by Matteo Collina<\/a>.
As you can understand, this solution only works in a Node.js environment.
Using this package, you can create one or more in-memory <\/strong>queues.
This means that when the process kills, you will lose all the data in the queue, and you will need a strategy to recreate the queue on the application’s startup.<\/p>\n\n\n\n

Getting started with fastq<\/h2>\n\n\n\n

To get started with fastq, you must install it on your Node.js project.
I have already created a simple project for you to follow along with the post; you can find it here<\/a>.
So, the first simple step, after the git clone<\/kbd><\/em> and the npm install<\/kbd><\/em>, is to run the command npm install fastq<\/kbd><\/em>, which installs the dependency in your project.
With that, you are ready to use fastq on your project.
As you can see, this package’s setup is smooth and easy to do.<\/p>\n\n\n\n

My first queue<\/h3>\n\n\n\n

Now, it’s time to create your first queue.
Fastq has a simple API to interact with it. You must create a function that is the worker for your queue. This function has the first argument, which is the item to handle in the queue, and a second argument, which depends on which approach you want to follow to complete the item’s execution. If you like to use the callback approach, the second argument is your done function to call at the end of the worker; if there is a successful execution, the callback must be invoked with null as the first parameter. Otherwise, you must pass the error as the first parameter. Instead, if you are a promise approacher, you can skip this argument and handle the error throwing it, as you already know.<\/p>\n\n\n\n

You’ll follow the promising approach for this post, but feel free to write the code using callback if you prefer.<\/p>\n\n\n\n

Let’s create a new file src\/queue.ts<\/code> containing our queue worker.
To simply this post, this queue will emulate the user’s sign-up in a real system, and its ownership will be to send a fake email to the user to confirm its email. As you can imagine, you won’t dive into all these details, but you’ll fake some services to have an accurate idea of what you can do with fastq.
So now it’s time to build your first queue! In the queue.ts<\/code> file, let’s start with the interface definition of the task.<\/p>\n\n\n

interface<\/span> UserCreatedTask {\n\u00a0id: string<\/span>;\n}<\/code><\/span>Code language:<\/span> TypeScript<\/span> (<\/span>typescript<\/span>)<\/span><\/small><\/pre>\n\n\n
This interface describes what your item in the queue looks like.
Now, you need to use it to handle the process in the queue, so first, you need to import some dependencies.<\/p>\n\n\n
import<\/span> type<\/span> { queueAsPromised } from<\/span> 'fastq'<\/span>;\nimport<\/span> * as<\/span> fastq from<\/span> 'fastq'<\/span>;\nimport<\/span> logger from<\/span> 'logger.js'<\/span>;\nimport<\/span> { setTimeout } from<\/span> 'node:timers\/promises'<\/span>;<\/code><\/span>Code language:<\/span> TypeScript<\/span> (<\/span>typescript<\/span>)<\/span><\/small><\/pre>\n\n\n
The first two imports are used for the queue; the logger shows some in the terminal during the execution, and the setTimeout fakes the execution time of an external service (in this case, a fake SMTP server).
With all these dependencies now you can create your worker.<\/p>\n\n\n
async<\/span> function<\/span> userInsertHandler<\/span>(arg: UserCreatedTask<\/span>) <\/span>{\n  logger.info(arg, 'User created task received'<\/span>);\n  const<\/span> fakeImplementation = Math<\/span>.random() > 0.8<\/span> ? 'success'<\/span> : 'error'<\/span>\n  const<\/span> timeout = fakeImplementation === 'success'<\/span> ? 2000<\/span> : 1000<\/span>;\n  await<\/span> setTimeout(timeout);\n  if<\/span> (fakeImplementation === 'error'<\/span>) {\n    throw<\/span> new<\/span> Error<\/span>(`User created task got error with id: ${arg.id}<\/span>`<\/span>);\n }<\/code><\/span>Code language:<\/span> TypeScript<\/span> (<\/span>typescript<\/span>)<\/span><\/small><\/pre>\n\n\n
As you can see, this function is easy but can give you all the ideas required to handle a queue with fastq.
First, the argument is of type UserCreatedTask<\/code>,  which means that you expect the item to have that shape.
Then, you faked a success or error result using a simple Math.random()<\/code>. If the result is successful, the execution will wait for 2 seconds and then resolve the promise; otherwise, the execution will wait for 1 second and raise an error.
Very easy, as you can understand.
But now you need to associate this worker to a real fastq queue; to do that, you must type this code.<\/p>\n\n\n
const<\/span> queue: queueAsPromised<UserCreatedTask> = fastq.promise(userInsertHandler, 1<\/span>);\n\nexport<\/span> default<\/span> queue;<\/code><\/span>Code language:<\/span> TypeScript<\/span> (<\/span>typescript<\/span>)<\/span><\/small><\/pre>\n\n\n
Using the fastq.promise<\/code> method, you create your queue with fastq and associate your userInsertHandler<\/code> worker to your queue. Last but not least, the second parameter is the number of concurrency workers you want to have; in this case, there is just one worker for this queue, but you can increase it if you prefer.
The last row exposes the queue outside this JS module and allows pushing items inside the queue.<\/p>\n\n\n\n
\"mcollina<\/figure>\n\n\n\n
Pushing data inside the code<\/h3>\n\n\n\n
The last puzzle piece is about pushing data inside the queue.
To do that, you need to import the queue in the consumer module and invoke the push method.
So, move to the src\/index.ts<\/code> file and import the queue.<\/p>\n\n\n
import<\/span> userInsertQueue from<\/span> '.\/queue.js'<\/span>;<\/code><\/span>Code language:<\/span> TypeScript<\/span> (<\/span>typescript<\/span>)<\/span><\/small><\/pre>\n\n\n
Now, you’ll create a fake running method that creates different user IDs and pushes them inside the queue.<\/p>\n\n\n
async<\/span> function<\/span> run<\/span>(<\/span>) <\/span>{\n  const<\/span> usersToHandle = new<\/span> Array<\/span>(5<\/span>).fill(undefined<\/span>).map(()<\/span> =><\/span> randomUUID())\n  for<\/span> (const<\/span> id of usersToHandle) {\n    const<\/span> task = {\n      id\n }\n    userInsertQueue.push(task)\n .then(()<\/span> =><\/span> {\n      logger.info(task, `Task with id ${task.id}<\/span> has been completed`<\/span>);\n })\n .catch((error<\/span><\/span>) =><\/span> {\n      logger.error(task, `Task with id ${task.id}<\/span> has failed`<\/span>, error);\n })\n    logger.info(task, `Task with id ${task.id}<\/span> has been pushed`<\/span>);\n }\n}\n\nrun()<\/code><\/span>Code language:<\/span> TypeScript<\/span> (<\/span>typescript<\/span>)<\/span><\/small><\/pre>\n\n\n
As you can see, the push method is a promise, so you can handle the promise result to understand when the task will be resolved or rejected.
Please remember this because it will be helpful when discussing the retry strategy.
Okay, let’s return to the example. In this run method, you create a user ID list using the randomUUID<\/code> method. Then, looping the list, you push the element inside the queue. Using the then and catch methods, you log the result of the execution of that ID.
Perfect, it’s time to check the result, so move to the terminal and run the command npm run start<\/kbd>.
You should see something like this in your terminal<\/p>\n\n\n
[14:35:13.067] INFO (59175): User created task received\n    id: \"a966759c-f385-4651-a0c1-55742ceb4017\"<\/span>\n[14:35:13.067] INFO (59175): Task with id a966759c-f385-4651-a0c1-55742ceb4017 has been pushed\n    id: \"a966759c-f385-4651-a0c1-55742ceb4017\"<\/span>\n[14:35:13.067] INFO (59175): Task with id 24d9d735-2399-471e-b64c-55ee0c769daa has been pushed\n    id: \"24d9d735-2399-471e-b64c-55ee0c769daa\"<\/span>\n[14:35:13.067] INFO (59175): Task with id 726b4bc5-794d-4581-a96c-e41a474c446d has been pushed\n    id: \"726b4bc5-794d-4581-a96c-e41a474c446d\"<\/span>\n[14:35:13.067] INFO (59175): Task with id 1a6dd1b4-2123-410b-a2ad-f420a14d53e9 has been pushed\n    id: \"1a6dd1b4-2123-410b-a2ad-f420a14d53e9\"<\/span>\n[14:35:13.067] INFO (59175): Task with id 6a8681e4-bba4-42e0-b484-976de8be9bcb has been pushed\n    id: \"6a8681e4-bba4-42e0-b484-976de8be9bcb\"<\/span>\n[14:35:14.069] INFO (59175): User created task received\n    id: \"24d9d735-2399-471e-b64c-55ee0c769daa\"<\/span>\n[14:35:14.070] ERROR (59175): Task with id a966759c-f385-4651-a0c1-55742ceb4017 has failed\n    id: \"a966759c-f385-4651-a0c1-55742ceb4017\"<\/span>\n[14:35:15.070] INFO (59175): User created task received\n    id: \"726b4bc5-794d-4581-a96c-e41a474c446d\"<\/span>\n[14:35:15.071] ERROR (59175): Task with id 24d9d735-2399-471e-b64c-55ee0c769daa has failed\n    id: \"24d9d735-2399-471e-b64c-55ee0c769daa\"<\/span>\n[14:35:16.071] INFO (59175): User created task received\n    id: \"1a6dd1b4-2123-410b-a2ad-f420a14d53e9\"<\/span>\n[14:35:16.071] ERROR (59175): Task with id 726b4bc5-794d-4581-a96c-e41a474c446d has failed\n    id: \"726b4bc5-794d-4581-a96c-e41a474c446d\"<\/span>\n[14:35:17.072] INFO (59175): User created task received\n    id: \"6a8681e4-bba4-42e0-b484-976de8be9bcb\"<\/span>\n[14:35:17.073] ERROR (59175): Task with id 1a6dd1b4-2123-410b-a2ad-f420a14d53e9 has failed\n    id: \"1a6dd1b4-2123-410b-a2ad-f420a14d53e9\"<\/span>\n[14:35:19.074] INFO (59175): Task with id 6a8681e4-bba4-42e0-b484-976de8be9bcb has been completed\n    id: \"6a8681e4-bba4-42e0-b484-976de8be9bcb\"<\/span><\/code><\/span>Code language:<\/span> Bash<\/span> (<\/span>bash<\/span>)<\/span><\/small><\/pre>\n\n\n
Double-checking the result, you can notice that some tasks have been failed and others have been completed. So, as you can understand, there isn’t any retry strategy out-of-the-box in fastq, because the failed tasks remain failed at the end of the execution.
So, how can you build your retry strategy?
In this example, a map can be used to save the task’s status and the number of retries.
Let’s see the implementations<\/p>\n\n\n
import<\/span> logger from<\/span> 'logger.js'<\/span>;\nimport<\/span> { randomUUID } from<\/span> 'node:crypto'<\/span>;\nimport<\/span> userInsertQueue from<\/span> '.\/queue.js'<\/span>;\n\ntype<\/span> Status = 'not-handle'<\/span> | 'error'<\/span> | 'success'<\/span>\n\nasync<\/span> function<\/span> run<\/span>(<\/span>) <\/span>{\n  const<\/span> usersToHandle = new<\/span> Array<\/span>(5<\/span>).fill(undefined<\/span>).map(()<\/span> =><\/span> randomUUID())\n  const<\/span> userProcessStatus: Record<string<\/span>, { status: Status, retryValue: number<\/span> }> = {}\n  for<\/span> (const<\/span> id of usersToHandle) {\n    userProcessStatus[id] = {\n      status: 'not-handle'<\/span>,\n      retryValue: 0<\/span>\n };\n    const<\/span> task = {\n      id\n }\n    const<\/span> pushMessage = ()<\/span> =><\/span> userInsertQueue.push(task)\n .then(()<\/span> =><\/span> {\n        logger.info(`Task with id ${task.id}<\/span> has been proceed`<\/span>);\n        userProcessStatus[id].status = 'success'<\/span>;\n })\n .catch(err<\/span> =><\/span> {\n        const<\/span> state = userProcessStatus[id]\n        logger.error(err, `Task with id ${task.id}<\/span> got error after retry ${state.retryValue}<\/span>`<\/span>);\n        state.status = 'error'<\/span>;\n        if<\/span> (state.retryValue < 3<\/span>) {\n          state.retryValue++\n          pushMessage()\n }\n })\n    pushMessage()\n    logger.info(task, `Task with id ${task.id}<\/span> has been pushed`<\/span>);\n }\n\n  process.on('exit'<\/span>, ()<\/span> =><\/span> {\n    for<\/span> (const<\/span> [id, { status, retryValue }] of Object<\/span>.entries(userProcessStatus)) {\n      logger.info(`Id ${id}<\/span> is ${status}<\/span> after ${retryValue}<\/span> retries`<\/span>);\n }\n })\n}\n\nrun()<\/code><\/span>Code language:<\/span> TypeScript<\/span> (<\/span>typescript<\/span>)<\/span><\/small><\/pre>\n\n\n
In this approach, the userProcessStatus<\/code> map contains the user IDs as keys and an object with two fields: the user’s status and the retries for that ID in the values.
Then, using the pushMessage<\/code> you create a method that handles this map. First, you push the element in the queue, and waiting for the result, you can handle the retry in case of failure. So, if the result succeeds, you can set the element in the map as completed. Still, if the element fails, you set the status as error<\/code>, and if the retry number is less than three, you increase the retry number and invoke the pushMessage again to retry that specific element.
I also added a hook before exiting the process to recap the tasks’ results just for debugging.
Now, you can run npm run start<\/kbd> again and look at the result. You will notice that the program tries to handle the failed tasks before ending.<\/p>\n\n\n\n
Conclusion<\/h2>\n\n\n\n
It’s time to wrap up this blog post.
In this article, you learnt:<\/p>\n\n\n\n