Using Observables in NestJS Microservices 🚀

NestJS is a progressive Node.js framework that heavily embraces Observables to handle asynchronous tasks. Observables are particularly useful in NestJS Microservices for:

✅ Inter-Service Communication (using Kafka, Redis, RabbitMQ, etc.)

✅ Streaming Data (WebSockets, gRPC, etc.)

✅ Handling Long-running Tasks (e.g., background jobs)

Let’s dive into real-world examples of how Observables can be leveraged in NestJS Microservices. 🔥

1️⃣ Observables in NestJS Microservices

NestJS uses RxJS Observables as a core part of its design for handling async operations. The framework provides built-in support for Microservices and encourages the use of Observables for request-response patterns.

Example: NestJS Microservices Setup

Let’s say we have two microservices:

• Orders Service (publishes an event when an order is placed)
• Inventory Service (listens and updates stock levels)

Orders Microservice (Publisher)

import { Controller, Post } from '@nestjs/common'; import { Client, ClientProxy } from '@nestjs/microservices'; import { Observable } from 'rxjs'; @Controller('orders') export class OrdersController { @Client({ transport: Transport.REDIS, options: { host: 'localhost', port: 6379 } }) private client: ClientProxy; @Post('create') createOrder(): Observable<string> { return this.client.send('order_created', { productId: 1, quantity: 2 }); } } 

🔹 Here, client.send() returns an Observable that emits a response when the Inventory Service processes the event.

Inventory Microservice (Listener)

import { Controller } from '@nestjs/common'; import { MessagePattern } from '@nestjs/microservices'; @Controller() export class InventoryController { @MessagePattern('order_created') handleOrderCreated(data: { productId: number; quantity: number }): Observable<string> { console.log('Updating inventory for:', data); return new Observable((subscriber) => { // Simulate processing setTimeout(() => { subscriber.next('Inventory Updated ✅'); subscriber.complete(); }, 2000); }); } } 

🔹 This service listens for order_created messages and responds with an Observable.

Response in Orders Service

createOrder().subscribe((response) => console.log(response)); 

🟢 Output:

Updating inventory for: { productId: 1, quantity: 2 } Inventory Updated ✅ 

2️⃣ Observables in HTTP Services

If you're building a REST API inside a microservice, Observables can be used with HTTP clients like Axios (wrapped in from() to convert Promises to Observables).

Example: Fetching Data from Another Microservice

import { Injectable } from '@nestjs/common'; import { HttpService } from '@nestjs/axios'; import { Observable, map } from 'rxjs'; @Injectable() export class ProductService { constructor(private httpService: HttpService) {} getProductDetails(productId: number): Observable<any> { return this.httpService.get(`http://inventory-service/products/${productId}`).pipe( map((response) => response.data) // Transform response ); } } 

🔹 This helps us keep the reactive approach even for HTTP calls.

3️⃣ Streaming Data Using Observables

NestJS supports WebSockets & gRPC, both of which work well with Observables.

Example: Real-time Stock Updates using WebSockets

1️⃣ Gateway (WebSocket Server)

import { WebSocketGateway, SubscribeMessage, WebSocketServer } from '@nestjs/websockets'; import { Observable, interval, map } from 'rxjs'; @WebSocketGateway() export class StockGateway { @WebSocketServer() server; @SubscribeMessage('stock_updates') stockUpdates(): Observable<{ stock: number }> { return interval(2000).pipe( map(() => ({ stock: Math.floor(Math.random() * 100) })) // Random stock value ); } } 

2️⃣ Client (WebSocket Frontend)

const socket = io('http://localhost:3000'); socket.emit('stock_updates'); socket.on('stock_updates', (data) => { console.log('Live Stock:', data); }); 

🔹 The stock updates stream continuously using an Observable interval.

4️⃣ Handling Long-running Tasks with Observables

In some cases, we might need to process large data asynchronously. Instead of blocking the request, we can return an Observable that emits data progressively.

Example: Streaming Large Report Data

import { Controller, Get } from '@nestjs/common'; import { Observable, interval, take, map } from 'rxjs'; @Controller('reports') export class ReportsController { @Get('generate') generateReport(): Observable<string> { return interval(1000).pipe( take(5), // Emit 5 values (simulate processing) map((count) => `Processing chunk ${count + 1}...`) ); } } 

🟢 Client Output (after hitting /reports/generate):

Processing chunk 1... Processing chunk 2... Processing chunk 3... Processing chunk 4... Processing chunk 5... 

🔹 This prevents blocking and streams responses incrementally.

Why Use Observables in NestJS? 🤔

✔ Better async handling – Unlike Promises, Observables allow multiple values over time.

✔ Reactive programming – Works great with real-time updates (WebSockets, Kafka, etc.).

✔ Powerful Operators – map(), filter(), mergeMap() make async transformations easier.

✔ Built-in NestJS Support – Microservices, WebSockets, and gRPC all use Observables by default.

Conclusion 🎯

Observables in NestJS shine when working with Microservices, real-time applications, and long-running processes. They provide a powerful way to handle asynchronous tasks efficiently, making your app more reactive and scalable.

🔥 If you’re building NestJS Microservices, start using Observables today! 🚀