Message Queues - the Empire's courier system

A legionary finishes registering, and the service sends him a welcome letter. The mail server happens to be down - and the registration fails. The man stands rejected at the gate, though his entry in the register succeeded without fault. A side matter failed and dragged the main one down with it.

Rome solved this differently. A commander does not wait for the messenger to return from Gaul - he leaves the letter in the courier box and returns to his affairs. The messenger will take it when he can; if the horse goes lame, the letter waits. In systems we call that box a message queue.

What a queue gives you

There is one advantage and it is the whole point: asynchronous communication. The producer - the one sending - does not wait for the consumer, the one receiving. It leaves the message and moves on.

Three consequences follow. Registration finishes at once, because it is no longer tied to the mail. A consumer's failure does not topple the producer - messages wait in the queue until it comes back. And when letters pile up, you add more consumers to the same queue without touching the producer.

The producer - leaving the letter

The sending application needs a client connected to a broker, the server that handles the queues:

1ClientsModule.register([
2  {
3    name: 'LEGION_SERVICE',
4    transport: Transport.RMQ,
5    options: {
6      urls: ['amqp://localhost:5672'],
7      queue: 'legion_queue',
8    },
9  },
10]);

Transport.RMQ selects RabbitMQ - the most widely used broker. urls is the server's address and queue the name of the box the letters go into. The name LEGION_SERVICE will serve to inject the client wherever it is needed.

emit versus send - two kinds of letter

With a client in hand, we send a message using one of two methods. The difference between them is fundamental:

1// emit - send and forget, we do not wait for a reply
2this.client.emit('legion.created', { id: 1, name: 'Legio X' });
3
4// send - send and wait for a reply
5const result = await this.client.send('legion.count', {}).toPromise();

emit() is fire-and-forget - you leave the letter and move on, expecting no answer. That is how you announce facts: a legion was created, a payment went through, a user registered. The sender does not care what the recipient does with it.

send() waits for a reply - it is a question, not an announcement. You use it when you need a result: how many legions there are, whether this identifier exists.

The choice between them decides whether you benefit from the queue at all. send() restores the waiting, and with it the coupling we were escaping - the consumer must be running again for the producer to move on. So for notifications, emails and logs always emit(); leave send() for cases where you genuinely need an answer.

The consumer - receiving the letter

On the other side of the queue stands a listening service:

1@Controller()
2export class LegionConsumer {
3  @EventPattern('legion.created')
4  async handleLegionCreated(@Payload() data: LegionCreatedDto) {
5    await this.mailService.sendWelcome(data);
6  }
7}

@EventPattern('legion.created') says: this method handles messages of that name. The name is a contract between producer and consumer - it must match character for character the one given to emit(). @Payload() extracts the message body, that is the object passed as the second argument.

Note that the consumer looks like an ordinary controller - because it works much the same way. The difference is where the request comes from: not from the HTTP network, but from the queue.

Acknowledgement - the receipt

The broker must know whether the letter arrived and was processed. Without that information it cannot delete it - and were it to delete on sending, a consumer crashing halfway through its work would mean the message is lost for good.

Hence the acknowledgement:

1@EventPattern('legion.created')
2async handleLegionCreated(@Payload() data: LegionCreatedDto, @Ctx() context: RmqContext) {
3  const channel = context.getChannelRef();
4  const originalMsg = context.getMessage();
5
6  try {
7    await this.mailService.sendWelcome(data);
8
9    channel.ack(originalMsg);
10  } catch (error) {
11    channel.nack(originalMsg, false, true);
12  }
13}

@Ctx() gives access to the RabbitMQ context. getChannelRef() returns the channel of communication with the broker, and getMessage() the raw message the acknowledgement concerns.

The heart of it is in the try block. channel.ack(originalMsg) confirms successful processing - only then does the broker remove the message from the queue. Note where that call sits: after the work is done, not before. Acknowledging at the top of the method would mean "I received it" rather than "I processed it", and a failure in sending the email would end in the letter being silently lost.

In catch sits nack - a refusal to acknowledge. Its last argument, true, tells the broker to put the message back in the queue to be tried later. This is exactly the resilience we introduced queues for: a failed attempt does not destroy the message.

Summary

The courier box works and the letters arrive:

• a message queue gives asynchronous communication - the producer does not wait for the consumer,
• so a consumer's failure does not topple the producer, and messages wait until the recipient is back,
• we configure the client with ClientsModule.register, Transport.RMQ, the broker address and a queue name,
• emit() is fire-and-forget - you announce a fact and expect no reply; send() waits for a reply and so restores the coupling,
• use emit() for notifications and emails; send() only when you genuinely need a result,
• the consumer listens with @EventPattern('name') and extracts the body with @Payload(); the name must match the one in emit(),
• channel.ack(originalMsg) confirms processing, and only then does the broker remove the message from the queue,
• acknowledge after the work, not before - otherwise a failure means a silently lost message,
• nack with a final true returns the message to the queue for another attempt,
• the full flow: the producer sends with emit → the message waits in the queue → the consumer receives it via @EventPattern → it acknowledges with channel.ack.

This is the module's last lesson. You can now check each part on its own, their cooperation, a request's whole road, measure coverage and endurance, and now also separate services so that one failure does not drag the rest down. For now remember: a queue is a courier box - the commander leaves the letter and returns to his affairs, and the receipt arrives only once the messenger truly got there.

1// Message Queues - Courier System of the Roman Empire
2import { Injectable } from '@nestjs/common';
3
4// 1. Simple message queue implementation
5class MessageQueue<T> {
6  private queue: Array<{ id: string; data: T; timestamp: Date }> = [];
7  private handlers: Map<string, (data: T) => Promise<void>> = new Map();
8
9  // Send message to queue
10  async publish(channel: string, data: T): Promise<string> {
11    const id = `msg-${Date.now()}`;
12    this.queue.push({ id, data, timestamp: new Date() });
13
14    const handler = this.handlers.get(channel);
15    if (handler) {
16      await handler(data);
17    }
18    return id;
19  }
20
21  // Subscribe to a channel
22  subscribe(channel: string, handler: (data: T) => Promise<void>) {
23    this.handlers.set(channel, handler);
24  }
25
26  getQueueStatus() {
27    return {
28      size: this.queue.length,
29      channels: Array.from(this.handlers.keys()),
30      oldestMessage: this.queue[0]?.timestamp,
31    };
32  }
33}
34
35// 2. NestJS Bull Queue - configuration
36// @Module({
37//   imports: [
38//     BullModule.forRoot({
39//       redis: { host: 'localhost', port: 6379 },
40//     }),
41//     BullModule.registerQueue({ name: 'tribute-collection' }),
42//   ],
43// })
44// export class QueueModule {}
45
46// 3. Producer - adding jobs to queue
47@Injectable()
48class TributeQueueService {
49  private queue = new MessageQueue<any>();
50
51  constructor() {
52    this.queue.subscribe('tribute', async (data) => {
53      console.log(
54        `Collecting tribute: ${data.province} - ${data.amount} gold`
55      );
56    });
57
58    this.queue.subscribe('alert', async (data) => {
59      console.log(`ALERT: ${data.message}`);
60    });
61  }
62
63  async scheduleTributeCollection(province: string, amount: number) {
64    return this.queue.publish('tribute', { province, amount });
65  }
66
67  async sendAlert(message: string, priority: 'low' | 'high') {
68    return this.queue.publish('alert', { message, priority });
69  }
70
71  getStatus() {
72    return this.queue.getQueueStatus();
73  }
74}
75
76// 4. Communication patterns:
77// - Point-to-Point: producer -> consumer
78// - Pub/Sub: producer -> many consumers
79// - Request/Reply: producer waits for response
80// - Fan-out: message to multiple queues
81
82// 5. Processor - processing jobs
83// @Processor('tribute-collection')
84// class TributeProcessor {
85//   @Process('collect')
86//   async handleCollect(job: Job<TributeData>) {
87//     const { province, amount } = job.data;
88//     return { collected: true, province };
89//   }
90//
91//   @OnQueueCompleted()
92//   onCompleted(job: Job, result: any) {
93//     console.log(`Job ${job.id} completed`);
94//   }
95// }
96