Implementing Real-Time Chat Messaging with NestJS, Server-Sent Events, and NestJS Event Emitters

Introduction

I recently implemented a real-time chat messaging feature using NestJS, Server-Sent Events (SSE), and NestJS event emitters. This combination of technologies allowed me to create an efficient, scalable, and easy-to-maintain solution. In this post, I will share my experience and provide a step-by-step guide on how to achieve this.

Server-Sent Events (SSE)

Server-Sent Events is a simple and efficient technology for sending real-time updates from the server to the client over a single HTTP connection. SSE allows the server to push events to the client without the need for constant polling or the overhead of WebSockets. SSE is built on top of the HTTP protocol, making it easy to implement and integrate with existing server frameworks like NestJS.

To implement SSE in your NestJS application, follow these steps:

1. Create a controller that handles SSE endpoints. Use the @Sse() decorator provided by NestJS to create the SSE endpoint. Clients can connect to this endpoint to receive real-time updates.

import { Controller, Sse } from '@nestjs/common';

@Controller('chats')
export class ChatController {
  @Sse(':chatId/messages')
  streamMessages(@Param('chatId') chatId: string) {
    // Logic to handle SSE stream
  }
}

1. In the streamMessages method of the ChatController, you can create an Observable that listens to the ‘messageCreated’ event from the MessageService. This event will be emitted whenever a new message is created.

import { Controller, Sse } from '@nestjs/common';
import { Observable } from 'rxjs';
import { map } from 'rxjs/operators';
import { MessageService } from './message.service';

@Controller('chats')
export class ChatController {
  constructor(private readonly messageService: MessageService) {}

  @Sse(':chatId/messages')
  streamMessages(@Param('chatId') chatId: string): Observable<Message> {
    return this.messageService.getMessageStream(chatId).pipe(
      map((message) => ({
        id: message.id,
        text: message.text,
        createdAt: message.createdAt,
      })),
    );
  }
}

1. The getMessageStream method in the MessageService should return an Observable that emits new messages whenever they are created. You can use NestJS event emitters to achieve this.

import { Injectable } from '@nestjs/common';
import { EventEmitter2 } from '@nestjs/event-emitter';
import { Observable } from 'rxjs';
import { Message } from './message.model';

@Injectable()
export class MessageService {
  constructor(private readonly eventEmitter: EventEmitter2) {}

  createMessage(chatId: string, text: string): Message {
    // Logic to create a new message
    const message: Message = {
      id: '...',
      chatId,
      text,
      createdAt: new Date(),
    };
    this.eventEmitter.emit('messageCreated', message);
    return message;
  }

  getMessageStream(chatId: string): Observable<Message> {
    return new Observable((subscriber) => {
      const listener = (message: Message) => subscriber.next(message);
      this.eventEmitter.on('messageCreated', listener);
      return () => this.eventEmitter.off('messageCreated', listener);
    });
  }
}

1. Finally, clients can connect to the SSE endpoint to receive real-time updates for a specific chat. Clients can connect to /chats/:chatId/messages to receive updates for the respective chat ID.

const eventSource = new EventSource('/chats/:chatId/messages');
eventSource.onmessage = (event) => {
  const message = JSON.parse(event.data);
  // Handle incoming message
};
eventSource.onerror = (error) => {
  // Handle SSE connection error
};

By following these steps, you can implement a real-time chat messaging feature using NestJS, Server-Sent Events, and NestJS event emitters. This approach provides an efficient and scalable solution for keeping clients updated with new messages in real-time.

I hope this guide helps you in implementing your own real-time chat messaging feature. Happy coding!