In this article, we will discuss some measures you can take to make systems much more efficient and reliable.

What is a Distributed System?

In a very simple way, a distributed system is a group or collection of independent computer nodes connected via a central unit which share and utilise computational resources.

Think of this as a network of systems which can interact and share tasks which they can process collectively as a single unit. Therefore distributed workloads refer to the processes or tasks that run in these systems.

Now, for these systems to interact efficiently, you can make use of some protocols or designs. One of them is a queue, which you'll learn about in the next section.

What is a Queue?

A queue is a data structure which models the First-In-first-Out Principle (FIFO).

To understand this, picture a checkout station in a supermarket. The first person to join the line will be the first to be attended to. Then subsequently, the next person is attended to. This is pretty much how queues works, but now as a data structure what is being processed can be different events or messages which we will identify as workloads.

Typically in a queue system, you have the producer and the consumer. The producer is the service that sends an event or in this context, a workload into the queue. And the consumer is the service that listens and processes it.

Streamlining Workflows: Producers Feed the Queue, Consumers Process the Flow.

With the very nature of these systems, running queues across this network can be demanding and sometimes you resort to some challenges which will be discussed briefly.

Challenges of Long-Running Queues in a Distributed System.

The goal of a task queue is primarily to handle asynchronous or long-running tasks efficiently.

However, within the distributed system context, running workloads across a network can sometimes prove to be challenging as they become overwhelming for these queues. This can leads to problems such as the following:

• Increase In Network Latency: Distributed systems primarily rely on communications across networks to efficiently coordinate workloads across the different computer nodes. Delayed or long-running tasks can lead to unpredictable response time(latency).
• Difficulty in tracking and monitoring: It can get very complex very quickly to track down each workload especially now that they are shared across different running systems in the network. In order to efficiently manage these you will have to make use of specialised tools such as the ELK Stack.
• Distributed Coordination: Ensuring that workloads being passed to different queues across the network are executed correctly and in the order they ought to is a challenge. Lack of this coordination further leads to data inconsistency.
• Fault Tolerance and Resilience: Processes running in a distributed system will surely contend with breakpoints and node failures. Having workloads in long-running queues can increase these failure points, which can lead to an increase in queue backlogs, data loss or inconsistency.

The question is how best can you combat these issues, or what measures should you take to reduce the possibility of system failures or data inconsistency. An approach to solving this is by making use of partitioned task queues.

What does it mean to partition a Queue?

Partitioning can be defined as breaking down a component into smaller subsections. Using this, you can derive a definition for queue partition as breaking a queue system into smaller, more manageable bits to offload workloads in a faster and more efficient manner.

Picture a centralised queue with various workloads coming with different task types or even priorities.

When traffic or workloads are at a very high rate, this queue will eventually become overwhelmed and keep compiling backlogs.

In order to lift the burden you can divide the queue into various bits to handle different task types. For example, you can have a queue to handle priority tasks or even to process different event types, like notifications.‌

Partitioned Queues in Distributed Systems: Optimizing Workload Management Across Nodes.

Primarily the goal of partitioning a queue is to improve system performance and reduce downtimes or data inconsistency. In order to do this, you have to look at different strategies of how to partition queues in other to know the one that is best suitable for your system.

Effective Task Queue Partitioning Strategies

There are various ways in which you can break a queue into subsections, but for you to be able to do that efficiently, you can employ some tested strategies such as Range-based Partitioning, Hash-based partitioning, and so on.

Hash-based partitioning

This form of partitioning allows for the selection of a queue using an assigned hash value.

First off, a parameter in the workload is used to get a hash value. Then, you take the modulo of that value with the number of partitions (queues), which results in a partition index. This index is then used to determine the queue to which the task will be assigned.

const numPartitions = 4;

const queues = Array.from({ length: numPartitions }, (_, i) => ({
    name: `Queue ${i + 1}`,
    partitionIndex: i + 1,
}));

function getPartitionIndex(hash) {
    const partitionIndex = hash % numPartitions;
    return partitionIndex + 1;
}

function hashCode(str) {
    let hash = 0;
    for (let i = 0; i < str.length; i++) {
        const char = str.charCodeAt(i);
        hash = ((hash << 5) - hash) + char;
    }
    return Math.abs(hash);
}

const tasks = [
    { id: 1, value: "Emails Task" },
    { id: 2, value: "Transactions Tasks"},
    { id: 3, value: "Orders Tasks" },
    { id: 4, value: "Update Tasks" },
];

tasks.forEach(task => {
    const hash = hashCode(task.value);
    const partitionIndex = getPartitionIndex(hash);
    const assignedQueue = queues.find(queue => queue.partitionIndex === partitionIndex);
    console.log(`Task with ID: ${task.id} (content: ${task.value}) assigned to ${assignedQueue ? assignedQueue.name : "no queue"}`);
});

The above example involves hashing a particular value or field in the workload or event that is to be assigned. In this case, the content field was hashed, and its hash value was then used to determine which queue the workload will be assigned to. This hash value might appear random but in truth it is constant, meaning the same input will always produce the same hash value.

Task with ID: 1 (content: Emails Task) assigned to Queue 3
Task with ID: 2 (content: Transactions Tasks) assigned to Queue 4
Task with ID: 3 (content: Orders Tasks) assigned to Queue 4
Task with ID: 4 (content: Update Tasks) assigned to Queue 2

Seeing the output, if you were to change the content of one of these tasks, you will see that they will get another hash value and might be assigned to a different queue.

In a real-life scenario, if you want the workloads to be distributed amongst different queues, you should ensure the hash value is something unique among tasks, like its Identifier. You can make use of an UUID.

Range-based Partitioning

This strategy allows you to spread tasks or workloads coming from a centralised workload into subsections based on a range of task IDs. For example, queue A might only allow a task ID that falls within the range of 1 - 50, while queue B handles the task with an ID ranging from 51- 100.‌

‌This strategy ensures that tasks are distributed across each partition in order to efficiently utilise the resources available and ensure workloads are distributed.

Take a look at this code example below that shows how tasks are assigned to a partition based on the range their IDs fall in:

const { v4: uuidv4 } = require("uuid");

const numPartitions = 4;
const partitionRanges = Array.from({ length: numPartitions }, (_, i) => ({
    start: i * 250,
    end: (i + 1) * 250,
}));

const queues = Array.from({ length: numPartitions }, (_, i) => ({
    name: `Queue ${i + 1}`,
    partitionIndex: i + 1,
}));

function getPartitionIndex(hash) {
    return Math.floor(hash / 250) + 1;
}

const hashCode = (str) => {
    let hash = 0;
    for (let i = 0; i < str.length; i++) {
        const char = str.charCodeAt(i);
        hash = (hash << 5) - hash + char;
    }
    return Math.abs(hash) % 1000;
};

const tasks = Array.from({ length: 4 }, () => ({ id: uuidv4() }));

tasks.forEach((task) => {
    const hash = Math.abs(hashCode(task.id));
    const partitionIndex = getPartitionIndex(hash);
    const assignedQueue = queues.find(
        (queue) => queue.partitionIndex === partitionIndex,
    );
    console.log(
        `Task with ID: ${task.id} assigned to ${assignedQueue ? assignedQueue.name : "no queue"}`,
    );
});

In this code example, the taskId is a UUID. To match them to a partitioned queue range (each queue has a range of 250), you first hash-code it and then convert it to an absolute value.

Next, you call the PartitionIndex() function with this hash value.

From there, It finds the partitioned queue that the taskId fits within the range.

This is an example output below:

Task with ID: 493fbf04-2f4f-43c0-9486-f5c99313d4e6 assigned to Queue 2
Task with ID: 9bad272f-3369-408e-bb37-5ef00c68b0b6 assigned to Queue 3
Task with ID: e0ec2d06-d66b-4e0e-8e85-04e4755a42be assigned to Queue 1
Task with ID: 960f415b-2e64-4169-b4ee-59b11c33b451 assigned to Queue 4

Round-Robin Partitioning

For this partitioning strategy, tasks are assigned to queues in a cyclic pattern. This means that if a task gets assigned to queue 1, the next task will get assigned to queue 2, until all the queues get assigned a task before starting from queue 1 again, therefore making it to be distributed evenly amongst the queues.

const numPartitions = 4;
let currentPartition = 0;

function getRoundRobinPartitionIndex() {
    const partitionIndex = (currentPartition % numPartitions) + 1;
    currentPartition++;
    return partitionIndex;
}

const tasks = [1, 2, 3, 4, 5]; 
tasks.forEach(task => {
    const partitionIndex = getRoundRobinPartitionIndex();
    console.log(`Task with ID: ${task} assigned to Queue ${partitionIndex}`);
});

For this example, a function getRoundRobinPartitionIndex() was defined and it determines the next queue index using a round-robin strategy.

It first calculates the queue index by taking the modular value of currentPartition and numPartitions. It adds 1 to it to ensure the initial index starts from 1, then increments the currentPartition to make sure the next queue to be assigned is greater than the previous. If the modular value is 0, it returns to the queue with index of 1.

Task with ID: 1 assigned to Queue 1
Task with ID: 2 assigned to Queue 2
Task with ID: 3 assigned to Queue 3
Task with ID: 4 assigned to Queue 4
Task with ID: 5 assigned to Queue 1

As you can see, the output above shows that the tasks were distributed to each of the queues and then started from the first queue again.

Dynamic Partitioning

This type of partitioning strategy involves assigning tasks to queues based on changing workloads or system conditions.

For instance, a system has 3 queues actively working and 1 that is free. It checks for the one with the least tasks assigned to it and then assigns a new task to it. By doing this, it ensures that the queues are not overwhelmed and that system resources are efficiently utilised.

const { v4: uuidv4 } = require('uuid');
let numPartitions = 4;
let queues = Array.from({ length: numPartitions }, (_, i) => ({
    name: `Queue ${i + 1}`,
    tasks: [],
}));

function dynamicPartitioning(task) {
    let minLoadQueue = queues[0];
    for (let queue of queues) {
        if (queue.tasks.length < minLoadQueue.tasks.length) {
            minLoadQueue = queue;
        }
    }
    minLoadQueue.tasks.push(task);
    console.log(`Task with ID: ${task.id} assigned to ${minLoadQueue.name}`);
}

const tasks = Array.from({ length: 5 }, () => ({ id: uuidv4(), value: `Task content` }));

tasks.forEach(task => {
    dynamicPartitioning(task);
});

function addPartition() {
    numPartitions++;
    queues.push({ name: `Queue ${numPartitions}`, tasks: [] });
    console.log(`Added new partition: Queue ${numPartitions}`);
}

setTimeout(() => {
    addPartition();
    // Assign new tasks to demonstrate dynamic partitioning with the new queue
    const newTasks = Array.from({ length: 5 }, () => ({ id: uuidv4(), value: `New task content` }));
    newTasks.forEach(task => {
        dynamicPartitioning(task);
    });
}, 5000);

In this code above, it starts by creating three queues and dynamically assign tasks to the least loaded queue.

After a simulated put-off, a new queue is brought to deal with the elevated workload and new obligations are distributed among all available queues.

This technique ensures that the burden is balanced throughout all queues, and the queue is overloaded, making the machine more efficient and responsive to adjustments in workload.

Task with ID: 396d4fde-830b-4443-b59b-c089f3a1db49 assigned to Queue 1
Task with ID: aa85af19-e46a-4b13-9a9c-2eb3ec535af0 assigned to Queue 2
Task with ID: f93d4cbb-0cad-4b71-a424-4c27a77ed38a assigned to Queue 3
Task with ID: c7e0cc7e-f055-4a38-9869-466406a33ca8 assigned to Queue 4
Task with ID: b49fe153-73e8-45d1-85b3-a5873f836dc9 assigned to Queue 1
Added new partition: Queue 5
Task with ID: a8a5f78e-da7e-4c65-9d74-8cef57f271f3 assigned to Queue 5
Task with ID: 00ce47b0-6098-49d6-ad06-820733ffe16e assigned to Queue 2
Task with ID: 7b1a5dcf-c42c-46dc-8e29-bae1cc5408c1 assigned to Queue 3
Task with ID: 3e31ad34-1a19-4d3c-a8f3-08f765a45f13 assigned to Queue 4
Task with ID: 73d55243-27da-44a4-9192-532a20889373 assigned to Queue 5

There are other strategies, like content-based partitioning, random-based partitioning, or even a user-defined custom type.

All these strategies are good but can still have some issues if not properly managed.

For instance, if you are using the range-based partition strategy and you select an identifier which falls within the set range of a queue, that queue partition will eventually get overwhelmed. It is left up to you to figure out the one that is best for your system requirements.

Conclusion

Processing workloads in distributed systems can be kind of complex and inefficient if not managed properly.

Partitioning queues across the system and having the workloads be distributed to them based on a strategy is one way to improve your system performance.

In this article, some challenges with long running queues were pointed out, and some strategies to relieve them were given. I have been researching distributed systems in run-time environments and databases and this was one of my findings.

I hope you enjoyed reading this because I enjoyed writing this. You can follow me on Twitter(X).

So you might wonder – how do you make sure your application runs without breaking? To handle this, you can send tasks to a queue outside your application's flow.

What is a Message Queue?

A message queue is a tool that facilitates the communication and transfer of data between services within a single application (or externally). It stores these data or messages using the First-In-First-Out (FIFO) principle. This means that older data that's passed into these queues gets processed before newer data.

Different components make up a message queue, such as:

• Messages: These are the data that are sent to the queue. They are often referred to as jobs.
• Queues: These are the data structures used for storing messages.
• Producers: These are a service that sends messages or data into the queue system.
• Consumers: These are a service that listens to the queue and executes messages passed in it.

Message Queuing Tools

Now, there are various message queuing tools you can use in asynchronous systems, like the following:

• RabbitMQ: a reliable and flexible option for implementing message queues in applications.
• Apache Kafka: an efficient message queuing tool, also very good at event stream processing.
• Redis: an in-memory store used for message queuing, caching, and data processing.

Note that some of these tools are not limited to message queuing but can be used for other purposes as well, like stream processing.

In this article, you will create a simple Nest.js project which will use RabbitMQ as the Message Queue Service Provider.

The tutorial will be divided into 3 parts:

• How to Set Up a Nest.js Project for Basic User Registration Flow
• How to Set Up an Email Service for User Registration
• How to Integrate a Queue Service using RabbitMQ

Prerequisites

• You'll need to have Node installed on your system. If you don't have it, here is its official site: https://nodejs.org/en.
• You'll need to have Node Package Manager (NPM) installed, which you can download here if you don't have it: https://docs.npmjs.com/downloading-and-installing-node-js-and-npm.
• You'll need to have installed RabbitMQ. Here is where you can get it in case you haven't yet: https://www.rabbitmq.com/download.html
• You'll need to have a text editor. For this article, I'll use VSCode. You can download it here: https://code.visualstudio.com/download or use the code editor of your choice.

How to Set Up a Nest.js Project

Spinning up a Nest.js application is fast and simple if you use the Nest CLI. Open up your terminal and enter this command below to install the CLI:

 $ npm install -g @nestjs/cli

This installs the Nest.js CLI globally on your system, meaning you can call the CLI commands regardless of the directory you are currently in.

Moving forward, to create a simple REST API project, you will enter the command below:

nest new simple-queue

Simple-queue here is the directory name that will be created. Inputting this command gives you a prompt to select a package manager.

When that's done, navigate to the created directory and open it in your text editor by entering this command:

cd simple-queue && code .

This opens up your text editor. We want to work on a project that will best show how a message queue can be used in a real-world scenario – so let's set up a basic user registration form. On successful data entry it sends an email to the user, but you'll handle the email service separately by passing it into a queue to improve performance.

For this, we'll be using an SQLite database, TypeOrm, class-validators, and the dotenv package so you can secure your config variables. Go ahead to install them by typing this command in your terminal:

npm install --save @nestjs/typeorm typeorm sqlite3 class-validator dotenv

When the installation is complete, go to your root app module, and then include the TypeOrm configuration for your database.

SQLite is a lightweight SQL database which allows us to quickly spin up and test data. It's optimal for this use case – and now we'll configure it.

Configuring the SQLite Database

import { Module } from "@nestjs/common";
import { TypeOrmModule } from "@nestjs/typeorm";
import { AppController } from "./app.controller";
import { AppService } from "./app.service";

@Module({
  imports: [
    TypeOrmModule.forRoot({
      type:'sqlite',
      database: 'mini-db.sqlite',
      entities: [__dirname + '/**/*.entity{.ts,.js}'],
      synchronize: true,  
  })],
  controllers: [AppController],
  providers: [AppService],
})
export class AppModule {}

Congrats! You have successfully connected a DB to your project. Now it's time to create the services that will handle the user registration.

In order to do this, you will have to go back to your dear friend the Nest CLI. There, you'll be inputting a different command to help generate a resource folder for the User, which will contain the entity, service, dto, and the controller.

To do this, open your terminal and enter in this command:

nest generate resource users

A prompt to select your transport layer will be shown. Select the first one which is the REST API. Then, another prompt will ask if you would like to generate CRUD endpoints – you can type Yes. Then you can make modifications according to your requirements.

To proceed, you first have to define what information each user should have. First, create a User entity. You can do this by navigating to the user entity file in the created entity subfolder in the user folder. Then define the user data like this:

import { Entity, PrimaryGeneratedColumn, Column } from "typeorm";

@Entity('users')
export class User {
  @PrimaryGeneratedColumn('uuid')
  id: string;

  @Column({ length: 100, unique: true })
  username: string;

  @Column({ length: 100, unique: true })
  email: string;
}

For this mini-project, you'll use basic user data to make the process faster. The username and email field have been set to be unique, meaning that there won't be a duplicate of the data instance passed in for this user table.

Now having done this, modify the create user dto file that was generated like this:

import { IsNotEmpty, IsString, IsEmail } from "class-validator";

export class CreateUserDto {
    @IsNotEmpty()
    @IsString()
    username: string;

    @IsNotEmpty()
    @IsString()
    @IsEmail()
    email: string;
  }

This was created to validate the payload that will be sent in your request by using the class-validator package.

Now, modify the create method in the user service file.

import { Injectable } from "@nestjs/common";
import { InjectRepository } from "@nestjs/typeorm";
import { Repository } from "typeorm";
import { CreateUserDto } from "./dto/create-user.dto";
import { User } from "./entities/user.entity";

@Injectable()
export class UsersService {
  constructor(
    @InjectRepository(User)
    private readonly userRepository: Repository<User>
  ) {}
  async create(createUserDto: CreateUserDto): Promise<User> {
    const newUser = this.userRepository.create(createUserDto);
    return await this.userRepository.save(newUser);
  }
}

Next you'll modify the controller file. You've already defined the create endpoint, so you'll just have to clean up the other endpoints that are not needed.

import { Controller, Post, Body } from "@nestjs/common";
import { CreateUserDto } from "./dto/create-user.dto";
import { UsersService } from "./users.service";

@Controller('users')
export class UsersController {
 constructor(private readonly usersService: UsersService) {}
  @Post()
  create(@Body() createUserDto: CreateUserDto) {
    return this.usersService.create(createUserDto);
  }
}

Open up the user module file and make some adjustments by adding the import field to the Module decorator and using the TypeOrmModule property.

import { Module } from '@nestjs/common';
import { UsersService } from './users.service';
import { UsersController } from './users.controller';
import { TypeOrmModule } from '@nestjs/typeorm';
import { User } from './entities/user.entity';

@Module({
  imports: [TypeOrmModule.forFeature([User])],
  controllers: [UsersController],
  providers: [UsersService],
})
export class UsersModule {}

Next, start up your server by entering this command on your terminal: npm run start:dev. Once the server is up and running, open up your API client of choice. For this article, we'll use Postman. Then make a POST request to the endpoint, which will be localhost:3000/users, providing the payload data required.

A request was made and a user instance was created.

Next up is adding an email service to your project which will help notify new users who are registering.

How to Set Up an Email Service for User Registration

For this, you'll use some packages which are required to create an email service. Open up your terminal and input the command below to install these packages:

npm install --save @nestjs-modules/mailer nodemailer

When these packages are installed, you can now implement the mail service. Using the Nest CLI, create a mailer module and service by entering this command in your terminal:

nest generate module email && nest generate service email

When it's done, open up the newly created module file in the mail folder. You'll use the MailerModule property of the @nestjs-modules/mailer package to configure your mail service here. It requires an SMTP client whose keys you'll need to configure this MailerModule.

For that you can use https://app.elasticemail.com to get these SMTP keys. Sign up and connect to the SMTP API. You'll then be given keys for your private use.

Note that this free mode of the SMTP client has limitations and it cannot send to all emails – so you should use a test email service.

How to Configure the Mailer Module

Once you have gotten that set up, go back to your application and create a .env file. Set your secrets for the SMTP keys. Then configure your MailerModule like this:

import { Global, Module } from "@nestjs/common";
import { EmailService } from "./email.service";
import { MailerModule } from "@nestjs-modules/mailer";

require('dotenv').config();
@Global()
@Module({
  imports: [
    MailerModule.forRoot({
      transport: {
        service: 'QueueTest',
        host: process.env.SMTP_HOST,
        port: process.env.SMTP_PORT,
        auth: {
          user: process.env.SMTP_USER,
          pass: process.env.SMTP_PASSWORD,
        },
      },
      defaults: {
        from: process.env.FROM_EMAIL,
      },
    }),
  ],
  providers: [EmailService]
})
export class EmailModule {}

The global decorator was set in order to make sure the MailModule can be called anywhere in your application. Make sure your secrets are properly loaded and thaqt you have a valid Email set in the from: process.env.FROM_EMAIL.

Check to make sure that the EmailModule is also imported in the root App Module the same way your UsersModule was imported in the Imports Array of the App Module.

Next, open your email service file – you'll need to make some modifications to the EmailService class. Add a constructor and call the MailService property from the @nestjs-modules/mailer package. Then go ahead and create a function that will handle sending the emails.

Below is a class and method that does this:

import { MailerService } from '@nestjs-modules/mailer';
import { HttpException, HttpStatus, Injectable } from '@nestjs/common';

@Injectable()
export class EmailService {
  constructor(private mailerService: MailerService) {}
  async sendEmail(options: { email: string; subject: string; html: string;
  }) {
    try {
      const message = {
        to: options.email,
        subject: options.subject,
        html: options.html
      };
      const emailSend = await this.mailerService.sendMail({
        ...message,
      });
      return emailSend;
    } catch (error) {
      throw new HttpException('Error', HttpStatus.INTERNAL_SERVER_ERROR);
    }
  }
}

Now you've defined the method to send an email. You've also put an exception handler in place for better error handling.

Now it's time to add this newly created service to your user registration flow.

Navigate to your user service file, and add the mail service to your constructor as a provider. Then call the service in your create user method like this:

@Injectable()
export class UsersService {
  constructor(
    @InjectRepository(User)
    private readonly userRepository: Repository<User>,
    private emailService: EmailService
  ) {}
  async create(createUserDto: CreateUserDto): Promise<User> {
    const newUser = this.userRepository.create(createUserDto);
    const user =  await this.userRepository.save(newUser);
      const emailData = {
        email: user.email,
        subject: 'Welcome to Our Community',
        html: `<p>Hello ${user.username},</p>
        <p>Welcome to our community! Your account is now active.</p>
        <p>Enjoy your time with us!</p>`,
      };
      await this.emailService.sendEmail(emailData)
    return user
  }
}

Make sure to modify your modules in order to correct any dependency injection errors. In your email module file, add the EmailService to the exports array:

 providers: [EmailService],
 exports: [EmailService]

Add it below your providers to export the Email Service so it can be accessed in other modules. Then import the EmailModule to your User module file and add it to your import array like this:

@Module({
  imports: [TypeOrmModule.forFeature([User]), EmailModule],
  controllers: [UsersController],
  providers: [UsersService],
})
export class UsersModule {}

Now it's time to test it. Get a free account from any online email testing platform and open Postman. Make a request to the create user endpoint with your valid email. You should get an email response like this:

Email response you should get

How to Integrate a Queue Service using RabbitMQ

To get started with this, you'll have to install some packages that let you implement queues using RabbitMQ. Enter the command below to install these packages:

npm install --save amqplib @types/amqplib amqp-connection-manager

Configure the Producer Service

Once installation is complete, it's time to configure RabbitMQ. You'll create a new folder in your src directory and name it queues. Then create the queue producer file. Import these packages and set them up like this:

import { HttpException, HttpStatus, Injectable, Logger } from '@nestjs/common';
import amqp, { ChannelWrapper } from 'amqp-connection-manager';
import { Channel } from 'amqplib';

@Injectable()
export class ProducerService {
  private channelWrapper: ChannelWrapper;
  constructor() {
    const connection = amqp.connect(['amqp://localhost']);
    this.channelWrapper = connection.createChannel({
      setup: (channel: Channel) => {
        return channel.assertQueue('emailQueue', { durable: true });
      },
    });
  }

  async addToEmailQueue(mail: any) {
    try {
      await this.channelWrapper.sendToQueue(
        'emailQueue',
        Buffer.from(JSON.stringify(mail)),
        {
          persistent: true,
        },
      );
      Logger.log('Sent To Queue');
    } catch (error) {
      throw new HttpException(
        'Error adding mail to queue',
        HttpStatus.INTERNAL_SERVER_ERROR,
      );
    }
  }
}

The AMQP connection was set and is running on localhost with the default RabittMQ port which is 5432. You also established a channel on that connection with an option input which is executed anytime a new channel is created. This helps if you have any configuration for that channel.

You also created an emailQueue with the assertQueue property which checks that a queue with that name does not already exist. If it does exist, it has no effect so it's idempotent.

Then you created an option durable: true to make sure that the queue will survive a server restart.

Next, you defined the method to add the email data to a queue. This calls the sendToQueue property of the channelWrapper, passing in the queue name you want to send the data to. Ideally, it should be the same name as the one you defined with the assertQueue property.

The second argument is the mail data, but firstly you converted it to a JSON string then to a Buffer. You do this because messages in RabbitMQ are mostly transmitted as binary data.

You can then set an option persistent: true to ensure that the data being sent to the queue won't be lost if the server crashes. Then with some error handling and the method to send messages to the queue, it's good to go.

Set Up the Consumer Service

Now that you've configured the producer service, it's time to set up the consumer service.

Create another file in the queue sub-folder. It's quite similar, but in this case, you will be consuming the data from the queue. Below is the configuration for the consumer service:

import { Injectable, OnModuleInit, Logger } from '@nestjs/common';
import amqp, { ChannelWrapper } from 'amqp-connection-manager';
import { ConfirmChannel } from 'amqplib';
import { EmailService } from 'src/email/email.service';

@Injectable()
export class ConsumerService implements OnModuleInit {
  private channelWrapper: ChannelWrapper;
  private readonly logger = new Logger(ConsumerService.name);
  constructor(private emailService: EmailService) {
    const connection = amqp.connect(['amqp://localhost']);
    this.channelWrapper = connection.createChannel();
  }

  public async onModuleInit() {
    try {
      await this.channelWrapper.addSetup(async (channel: ConfirmChannel) => {
        await channel.assertQueue('emailQueue', { durable: true });
        await channel.consume('emailQueue', async (message) => {
          if (message) {
            const content = JSON.parse(message.content.toString());
            this.logger.log('Received message:', content);
            await this.emailService.sendEmail(content);
            channel.ack(message);
          }
        });
      });
      this.logger.log('Consumer service started and listening for messages.');
    } catch (err) {
      this.logger.error('Error starting the consumer:', err);
    }
  }
}

First, you defined your consumer class. For this, it implements the onModuleInit interface which is provided by @nestJs/common. This specifies that the defined class should have a method named onModuleInit().

Like the name says, the method will be called automatically during the module initialization which is when the module containing this class is loaded.

In the class constructor, you added the emailService because you'll be using the sendEmail method of that class.

In the onModuleInit() method, you defined a channel. This is necessary because you need a channel to consume messages from a queue.

From this, the channel is then used to assert a queue which should be similar in name and options to what you have on your producer service. If it's not, you won't be able to listen to the queue created on the producer service.

Then you used the consume method of channel to listen and execute the message coming from the queue you have registered.

Recall that before, you had to convert the message to Buffer in order to send it into a queue. Now, you have to convert it to a JavaScript object. Then call the emailService method to send an email and pass in the converted JavaScript object as the argument of that method.

Finally, you called the ack method which is used to inform the queue that the message has been received and processed successfully in order for it to be removed from the queue.

Now that you've defined these services, create a module file and set them in the providers array. Then export the producer service because you will be calling it in another module.

import { Module } from '@nestjs/common';
import { ConsumerService } from './consumer.file';
import { ProducerService } from './producer.file';

@Module({
  providers: [ProducerService, ConsumerService],
  exports: [ProducerService],
})
export class QueueModule {}

Next up is to add the emails being sent on user registration to the queue service that you just created.

Navigate back to your user service file and make some modifications: replace the email service with the producer service as a provider in the constructor, and then call the service and the method to add to the email queue as shown below:

import { Injectable } from "@nestjs/common";
import { InjectRepository } from "@nestjs/typeorm";
import { Repository } from "typeorm";
import { CreateUserDto } from "./dto/create-user.dto";
import { User } from "./entities/user.entity";
import { ProducerService } from "src/queues/producer.file";

@Injectable()
export class UsersService {
  constructor(
    @InjectRepository(User)
    private readonly userRepository: Repository<User>,
    private producerService: ProducerService,
  ) {}
  async create(createUserDto: CreateUserDto): Promise<User> {
    const newUser = this.userRepository.create(createUserDto);
    const user = await this.userRepository.save(newUser);
    const emailData = {
      email: user.email,
      subject: 'Welcome to Our Community',
      html: `<p>Hello ${user.username},</p>
        <p>Welcome to our community! Your account is now active.</p>
        <p>Enjoy your time with us!</p>`,
    };
    await this.producerService.addToEmailQueue(emailData);
    return user;
  }
}

Also in the user module file, replace the EmailModule with that of the QueueModule to avoid dependency injection errors when you start up your server.

import { Module } from '@nestjs/common';
import { UsersService } from './users.service';
import { UsersController } from './users.controller';
import { TypeOrmModule } from '@nestjs/typeorm';
import { User } from './entities/user.entity';
import { QueueModule } from 'src/queues/queue.module';

@Module({
  imports: [TypeOrmModule.forFeature([User]), QueueModule],
  controllers: [UsersController],
  providers: [UsersService],
})
export class UsersModule {}

Now finally, it's time to test the user registration flow again. So navigate back to Postman and then type in a valid email and username and hit enter. On the terminal of your server running, you will see logs that were set in order to track the way the message got sent and how it was received and executed.

Logs that help you track the message

You can also open up the RabbitMQ dashboard to view queue activity on http://localhost:15672, By default the user is "guest", so enter in guest for the username and password.

RabbitMQ Queues and Streams

Here's the link to the GitHub repository. Feel free to check it out whenever you're stuck.

Conclusion

In this article you learned what a message queue is along with some major components of how they work. You also built a mini Nest.js project and implemented an email service in it. Finally, you integrated the queue service into your project, showing how it works in a real-life scenario.

Understanding message queue behaviors and patterns is an essential skill when developing scalable applications. This helps reduce lag and improves the speed and efficiency of your applications.

I hope you enjoyed reading this article. You can follow me on Twitter.

You might ask, what is Java? You may also wonder what it means for a programming language to be interpreted or compiled. Let's explore these concepts in the following sections.

What is the Java Programming Language?

Java is a programming language that is platform-independent and adopts the Object Oriented Programming structure to build software. It is not only limited to the web but also mobile and desktop.

You might ask what it means to be platform-independent. This is about the capability of Java programs to effortlessly operate on various operating systems like Windows, Mac, and Linux.

Unlike languages like C or C++, which can be limited to a solitary Operating System, Java's adaptability eliminates the limitation of depending on a specific operating system for it to operate.

To understand Java better, let's take a look into its history and the purpose it was created for.

A Brief History Of Java

In 1991, Java was developed by James Gosling and his team at Sun Microsystems. The first version of the language, 1.0, was released in 1996.

It was designed to have similar syntax to other higher-level languages like C and C++, so that engineers at that time wouldn't have a problem understanding and developing a program with Java.

It was developed with the intent to develop television interactivity, but it was too advanced for its time. The technology infrastructure and devices of that time were not adequately prepared to leverage Java's advanced features.

In recent times, we have seen Java still in the spotlight as it is the software that most big tech companies adopt due to its versatility. It's very much in demand as a primary language for building enterprise-level applications on different platforms.

Moving forward in our exploration of Java's inner workings, there are some key concepts you need to understand. Those concepts are understanding what it means for a programming language to be interpreted or compiled.

Interpreted vs Compiled Languages

We have often heard of these terms, but to understand them better, let us picture a foreign student embarking on two journeys to different locations.

In the first scenario, they arrive at the location without knowing the local language or customs, so they rely on an interpreter or translator to be able to communicate with the natives of that area. This is much like how an interpreter in an interpreted programming language translates code on-the-fly for the computer to understand.

In the second scenario, the student had the foresight to partake in classes to learn the language they will be needing, and is familiar with the customs of the location they will be going to. On arrival, they doesn't need an interpreter in order to communicate with the natives of that area. This is similar to how a compiled language functions.

By understanding these two scenarios, you can spot the differences between an interpreted language and a compiled language.

The former translates each line of code on the go. That is a real-time translation, much like how the student in the example made use of a translator/interpreter.

On the other hand, the latter first predigests the code to ensure smooth flow. Then it 'compiles' or transforms it into machine code which the computer can understand and execute, similar to our prepared student communicating fluently.

Each of these programming terms has a similar purpose of allowing communication between computer systems and the programmers.

Now, as you inquire deeper into these programming terms, let's understand how Java works.

How does Java Work?

Java is unique in the sense that it is both compiled and interpreted.

The basic flow is that it is first compiled from source code into what we call bytecode. Then, it is interpreted to machine code using an interpreter named JVM, which stands for Java Virtual Machine. This interpretation happens at runtime, so you can refer to Java as an interpreted language and also a compiled language.

Now you might ask what is bytecode?

The common terms we hear in program compilation and interpretation are mostly source code and machine code, The bytecode is akin to machine code in the sense that is it a compiled source code, but it's not executable unless interpreted by a virtual machine, in this case, the Java Virtual Machine.

What is a Java Virtual Machine?

The JVM (Java Virtual Machine) is a platform dependent virtual machine that interprets and executes Java Programs in bytecode and any other program from other languages that compiles to Java bytecode.

Java bytecode has a special quality: it's not tied to a specific platform.

Java bytecode, which is akin to a universal dialect, can be executed by a Java Virtual Machine (JVM) on most operating systems you can envision. Presently, when the JVM starts its processing, it converts that bytecode into machine code that's an exact match for the operating system it's running on.

But here's where it gets interesting: while Java bytecode is like a chameleon that can adapt to different platforms, the JVM itself isn't quite as flexible. It's like having different versions of a tool, each custom-made for a specific operating system.

So, you've got different flavours of JVMs to match the variety of operating systems out there.

Now you might be wondering since Java undergoes two processes, compilation and interpretation, does that make it slow? The answer is no. Although it undergoes various processing, it still is fast.

To understand this, we need to discuss some fundamentals seen in the JVM such as the JIT (Just In Time Compilation), Hotspot JVM, and its memory/storage uniqueness.

How does Java Attain High Speed?

It is important to note that Java streamlined its compilation for fast execution and took the platform independence of an interpreted language.

Compiled languages are known for their speed in processing due to them converting the source code fully and not line by line, unlike their counterpart the interpreted languages. Java compiles to bytecode which is faster to interpret using the JVM due to bytecode being very compact and optimized already.

JVM has several components such as its ability to manipulate memory at run time. Due to automatic memory management, less time is spent on the allocation of memory size for data using its inbuilt garbage collector.

How to make use of the JIT Compiler to achieve faster execution

Another interesting feature in the JVM, which helps to cut down time spent on improving code processing speed, is the JIT, which is an acronym for Just In Time.

The JIT compiler is at the core of the execution engine in the JVM. It observes as bytecode is being interpreted and executed, and stores a "Hotspot". This is a repeated set of executed bytecode.

Instead of reinterpreting and executing this same bytecode again, the JIT compiler stores this frequently executed bytecode in memory as a native machine code for future use.

This cached machine code is what gets executed when you run your bytecode, which results to a faster execution time. This whole process can be called a dynamic compilation.

What is garbage collection and how it improves JVM performance?

The JVM is unique in the way that it automates memory management. This involves memory allocation for programs it executes.

Unlike C/C++, where you write a code and then manually compute the memory allocation of the program, this feature improves and enhances the productivity of the programmers. It also reduces some errors that can result due to running out of memory or memory leaks.

Garbage collection aids this by automatically reclaiming memory occupied by objects in a program that is no longer in use. Doing so, frees up that memory for future use.

You might ask how this works. It follows the steps outlined below:

• It first identifies unreachable objects. These are objects in a program that are not in use. It's strict in doing this because if it selects an object in use and reclaims it, it will crash our program because there will be a referenced object error. That is a term called dangling references.
• Next, it reclaims the memory of the selected unused objects, which frees up memory for other future objects.
• In some cases, during this collection, the garbage collector compacts the remaining objects in memory, which improves memory locality. This is optional, as it depends on the garbage collection algorithms in use.

This is a rundown of how the garbage collector process works in the Java Virtual Machine.

Depending on the set garbage collector algorithm in place, it is important to note that the collection is triggered by the JVM when it detects low memory space, and kickstarts the steps above.

It is also essential to note that sometimes a pause in execution will happen in other to memory space.

ByteCode Optimization

There are some extra techniques performed on the compiler which help enhance performance.

The Compiler optimizes the bytecode by taking note of similar calculations. Instead of rerunning these calculations, it just outputs the result. This is called constant folding.

It also removes codes out of scope or codes that will never be executed. Doing this, eliminates unnecessary computation.

Another technique used by the compiler is the peephole optimization, which involves the replacement of bytecode instructions with a more efficient sequence.

So many factors contribute to the performance and speed of Java Programs.

These range from it both being compiled and interpreted, to the intricacies of some components in its interpreter (the Java Virtual Machine). All these play a major role in achieving high speed.

Ultimately, to have faster programs the programmer must write cleaner and organized code using good data structures.

Conclusion

In this article, you learned about a brief history of Java, and the differences between compiled and interpreted languages. You also learn about and some of Java's major components, such as the Java Virtual Machine, and how it contributes to the performance of Java programs.

If you enjoyed reading this article, you can share and follow me on Twitter and Linkedin.

And even though errors and bugs are a part of life, they can stress you out and decrease your productivity. Fortunately, you can limit the number of pests in your code by taking proactive measures to prevent and fix them.

In this article, you will learn how best to utilize exception filters to limit disruptions in your code implementations.

What Are Exception Filters?

Exception filters are constructs of a programming language that help you handle exceptions or errors found in services or controller classes. Using these filters improves the efficiency of your codebase and makes errors more traceable, which helps you resolve them.

What are Exception Filters in Nest.js?

Nest.js has an inbuilt exception filter interface that you import from the @nestjs/common package. This interface gives you precise information about exceptions you may encounter so that you can know how to fix them.

Some built-in exception filters include NotFoundException, UnauthorizedException, and RequestTimeOutException, among others.

How to Create a Custom Exception Filter

To create a custom exception filter class, you define it with a @Catch decorator that takes in the type of exception the filter should catch. This class then implements the ExceptionFilter interface. This way you will have access to the catch method that the interface provides.

NestJS allows you to create your own exception filters so you can define what sort of information it should return.

@Catch(HttpException)
export class TestExceptionFilter implements ExceptionFilter {
  catch(exception: HttpException, host: ArgumentsHost) {
    const ctx = host.switchToHttp();
    const response = ctx.getResponse<Response>();
    const request = ctx.getRequest<Request>();
    const status = exception.getStatus();

    response
      .status(status)
      .json({
        statusCode: status,
        timestamp: new Date().toISOString(),
        path: request.url,
      });
  }
}

As you can see above, the catch method takes in two argument inputs: the exception:HttpException and the host:argumentsHost.

HttpException is the exception thrown when an HTTP request fails and returns the appropriate Error Message with a status code as a response to the client.

The argumentsHost parameter provides information about the current request and response cycle. Here the code extracts the Response and Request objects from the ArgumentsHost object using the switchToHttp method. It then calls the getStatus method from the HttpException object in order to retrieve the HTTP status code of the error.

A JSON object gets returned. It contains the status code, the error timestamp, and the request URL, setting the HTTP response status code to be the error status code.

Another exciting thing to note is that you can modify the status code and error message of existing exception filters like NotFoundException so that they display your own custom error message and status code. Here is an illustration below:

export class NotFoundException extends HttpException {
  constructor(message: string) {
    super(message || 'Not Found', HttpStatus.NOT_FOUND);
  }
}

The example above shows that the NotFoundException class was modified to accept a message parameter in the constructor method or give a default value of 'Not Found'.

By doing this you are now able to customize an error message for your NotFoundException class and return a 404 status code.

How to Bind Exception Filters in Nest.js

You might be wondering how you can implement the custom or modified exception you've created for your application. Well, you use something called binding.

There are several ways in which you can bind exception filters to your application. Here are three major ways you can do it:

1. Global Scope
2. Controller Scope
3. Method Scope

Binding using Global Scope

Just like the name “Global” implies, it covers the entirety of a thing – in this case the application. The custom exception filter is bonded to the entirety of your web application by using the useGlobalFilters() function. An instance of your custom exception filter gets passed in before starting up the server.

Here is an example of what that looks like:

async function bootstrap() {
  const app = await NestFactory.create(AppModule);
  app.useGlobalFilters(new TestExceptionFilter());
  await app.listen(3000);
}
bootstrap();

If you bind the exception filter to the global scope, it means that any exceptions of type HttpException thrown from any controller or method within the application will be handled by the TestExceptionFilter.

Binding using Controller Scope

If you want to handle an exception that not only covers the specific methods in your application but also the entire controller with the methods in it, you can the @UseFilters to bind the custom exception to the controller. Check out the below is example:

@UseFilters(TestExceptionFilter)
@Controller('users')
export class UsersController {
  @Get()
  async findUsers() {
    // ...
  }
}

In this example, the custom exception filter is bonded to the controller by putting it ahead of the @Controller decorator. This way it handles any error that comes from the routes in this controller.

Binding using Method Scope

In this form of binding, you make use of the @UseFilters decorator to apply the Custom ExceptionFilter you defined to a method in your application. Here is an example:

@Controller()
export class UsersController {
  @Get()
  @UseFilters(TestExceptionFilter)
  async findUsers() {
    // ...
  }
}

In this example, you're using the @UseFilters decorator to bind the TestExceptionFilter exception filter to the findUsers method. This means that if any exception occurs within the findUsers() method, the TestExceptionFilter will catch and handle it.

Debugging Made Easier

In this tutorial, you learned about exception filters in Nest.js, and how to create custom exception filters. You also saw some different ways in which you can implement them in your applications.

All the methods discussed are good ways to handle exceptions. By implementing these techniques, you can ensure smoother application performance and provide a better user experience.

If you enjoyed reading this article, you can share and follow me on Twitter and Linkedin.