RabbitMQ RPC pattern in TypeScript

Services talk to each other in two common shapes: Request/Response and Event-Driven. Event-Driven communication is asynchronous and does not wait for an answer. The Request/Response pattern, implemented through Remote Procedure Calls (RPC), sends a call and waits for the response to come back. RabbitMQ carries both shapes, and this post implements the second one end to end.

RabbitMQ carries the messages between the two services

RabbitMQ is an open-source message broker software that enables applications to communicate with each other and share data by sending messages. It supports multiple messaging protocols, message queuing, delivery acknowledgement, and flexible routing to queues, making it highly reliable and scalable for modern application architectures. If you're not familiar with how exchanges and queues fit together, this breakdown of RabbitMQ exchanges vs queues is a good starting point before diving into the implementation.

RPC makes a remote call look like a local function call

Remote Procedure Calls (RPC) are a protocol that one program can use to request a service from a program located on another computer in a network without needing to understand network details. RPC abstracts the communication, so developers can call functions on remote servers just as they would do on a local system, expecting a response back.

RPC over RabbitMQ costs you an extra hop compared to a direct REST or gRPC call, because the request and the reply both travel through the broker. You pay that hop to get queueing, backpressure, and a responder you can scale on its own.

An inventory check needs an answer before the cart can continue

Take an e-commerce application where a user adds an item to their cart. Before confirming the purchase, the application needs to verify if the item is in stock. RPC with RabbitMQ handles that in five steps:

1. Client (Cart Service): The cart service prepares an RPC request containing the item ID and sends it to a dedicated RabbitMQ queue.
2. Server (Inventory Service): A separate inventory service listens on that queue for incoming requests.
3. Inventory Check: Upon receiving the request, the inventory service checks its database for the item's availability.
4. Response: The inventory service sends an RPC response back to the cart service, indicating whether the item is in stock or not.
5. Cart Update: Based on the response, the cart service can update the user interface and potentially offer alternatives if the item is unavailable.

The service split here is simplified, but the scenario shows the three benefits of RPC:

• Decoupling: The cart service doesn't need to know the internal workings of the inventory service. They communicate solely through messages.
• Scalability: The inventory service can be scaled independently to handle high traffic without affecting the cart service.
• Resilience: If the inventory service is temporarily unavailable, the cart service can handle the fallback gracefully.

Implementing RabbitMQ RPC in TypeScript

Both sides of this TypeScript RPC call use the same amqplib client on Node.js.

The cart service sends the request and waits on a reply queue

The client declares an exclusive queue for the answer, tags the request with a correlation ID, and resolves a promise when a reply carrying that same ID arrives.

import { connect } from "amqplib";
import { randomUUID } from "crypto";
 
// Function to call the inventory service
async function checkInventory(itemId: string): Promise<boolean> {
	// Create a connection to RabbitMQ
	const connection = await connect("amqp://localhost");
	const channel = await connection.createChannel();
 
	// Declare a temporary queue for responses
	const replyQueue = await channel.assertQueue("", { exclusive: true });
 
	// Generate a unique correlation ID
	const correlationId = randomUUID();
 
	// Prepare the RPC request message
	const message = {
		itemId,
		replyTo: replyQueue.queue,
		correlationId,
	};
 
	// Send the request to the RPC queue
	await channel.sendToQueue(
		"inventory_checks",
		Buffer.from(JSON.stringify(message)),
		{
			correlationId,
			replyTo: replyQueue.queue,
		},
	);
 
	// Make sure to disconnect the channel
 
	// Consume responses from the temporary queue
	return new Promise<boolean>((resolve) => {
		channel.consume(replyQueue.queue, (msg) => {
			if (msg.properties.correlationId === correlationId) {
				channel.ack(msg);
				resolve(JSON.parse(msg.content.toString()).inStock);
			}
		});
	});
}

Notice that the promise has no timeout. If the inventory service never replies, nothing resolves and the connection stays open, so add a timeout and close the channel before you ship this.

The inventory service replies to the queue named in replyTo

The server reads the request off the shared queue, does the lookup, and sends the answer to the reply queue and correlation ID that came in on the message properties.

import { connect } from "amqplib";
 
async function listenForInventoryChecks() {
	// Create a connection to RabbitMQ
	const connection = await connect("amqp://localhost");
	const channel = await connection.createChannel();
 
	// Declare the RPC queue
	await channel.assertQueue("inventory_checks");
 
	// Consume messages from the RPC queue
	channel.consume("inventory_checks", async (msg) => {
		const request = JSON.parse(msg.content.toString());
		const itemId = request.itemId;
 
		// Check inventory for the requested item
		const inStock = await checkInventoryDatabase(itemId);
 
		// Prepare the response message
		const response = { inStock };
 
		// Send the response back to the client's temporary queue
		await channel.sendToQueue(
			msg.properties.replyTo,
			Buffer.from(JSON.stringify(response)),
			{
				correlationId: msg.properties.correlationId,
			},
		);
 
		channel.ack(msg);
	});
}