I’m currently working on integrating RabbitMQ into my monolithic NestJS application for real-time inventory management as part of my e-commerce app. I want to use a fanout exchange to broadcast stock updates to multiple queues, such as an email queue and a log queue. However, I’m facing some issues with my current implementation.

Below are all the relevant code pieces in detail. Although the app is not designed as microservices, I expect it to act so, maintaining communication between services through RabbitMQ. My goal is to emit the pattern from inventory.service to the exchange and then fan out the messages to both queues, which are email_queue and log_queue.Going for just one queue worked pretty nice but I dont want to go with this option since that will cause some performance issues, that’s why I’m on seperate queue for each service that will listen the pattern

the workflow should be simply something like that:

here is my current implementation:

.env

RABBIT_MQ_EMAIL_QUEUE=stock_update_email_queue
RABBIT_MQ_LOG_QUEUE=stock_update_log_queue

rabbitmq.module.ts

import { DynamicModule, Module } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { ClientsModule, Transport } from '@nestjs/microservices';
import { RabbitMQService } from './rabbitmq.service';

interface RmqModuleOptions {
  name: string;
}

@Module({
  providers: [RabbitMQService],
  exports: [RabbitMQService],
})
export class RmqModule {
  static register({ name }: RmqModuleOptions): DynamicModule {
    return {
      module: RmqModule,
      imports: [
        ClientsModule.registerAsync([
          {
            name,
            useFactory: (configService: ConfigService) => ({
              transport: Transport.RMQ,
              options: {
                urls: [configService.get<string>('RABBIT_MQ_URI')],
                queue: configService.get<string>(`RABBIT_MQ_${name.toUpperCase()}_QUEUE`),
                queueOptions: {
                  durable: true,
                },
              },
            }),
            inject: [ConfigService],
          },
        ]),
      ],
      exports: [ClientsModule],
    };
  }
}

rabbitmq.service.ts

import { Injectable, Logger } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { RmqOptions, Transport } from '@nestjs/microservices';

@Injectable()
export class RabbitMQService {
  private readonly logger = new Logger(RabbitMQService.name);
  constructor(private readonly configService: ConfigService) {
    this.logger.log('RabbitMQService initialized');
  }

  getOptions(queue: string): RmqOptions {
    return {
      transport: Transport.RMQ,
      options: {
        urls: [this.configService.get<string>('RABBIT_MQ_URI')],
        queue: this.configService.get<string>(
          `RABBIT_MQ_${queue.toUpperCase()}_QUEUE`,
        ),
     
      },
    };
  }
}

inventory.module.ts

import { Module, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
import { InventoryService } from './inventory.service';
import { InventoryController } from './inventory.controller';
import { AccessModule } from '@app/common/access-control/access.module';
import { RedisModule } from '@app/common/redis/redis.module';
import { DatabaseModule } from 'src/database/database.module';
import { JwtService } from '@nestjs/jwt';
import { ProductModule } from 'src/product/product.module';
import { RmqModule } from '@app/common/rabbit-mq/rabbitmq.module';
import { AmqpConnection } from '@nestjs-plus/rabbitmq';
import { EmailModule } from 'src/email/email.module';

@Module({
  imports: [
    AccessModule,
    RedisModule,
    DatabaseModule,
    ProductModule,
    EmailModule,
    RmqModule.register({
      name: 'inventory',
    }),
  ],
  providers: [InventoryService, JwtService, AmqpConnection],
  controllers: [InventoryController],
})
export class InventoryModule {}
**your text**

inventory.service.ts

import { Injectable, Logger, Inject } from '@nestjs/common';
import { Prisma } from '@prisma/client';
import { DatabaseService } from 'src/database/database.service';
import { RedisService } from '@app/common/redis/redis.service';
import { Product } from '@prisma/client';
import { Variant } from '@prisma/client';
import { ProductService } from 'src/product/product.service';
import {
  NotFoundException,
  InternalServerErrorException,
} from '@nestjs/common';
import { ClientProxy } from '@nestjs/microservices';
@Injectable()
export class InventoryService {
  private readonly logger = new Logger(InventoryService.name);
  constructor(
    private readonly databaseService: DatabaseService,
    private readonly productService: ProductService,
    @Inject('RABBITMQ_CLIENT') private readonly client: ClientProxy,
  ) {}

  async updateProductStock(
    productId: string,
    quantity: number,
  ): Promise<Product> {
    try {
      const product = await this.productService.getProductById(productId);
      if (!product) {
        throw new NotFoundException('Product not found');
      }

      const updatedProduct = await this.databaseService.product.update({
        where: { id: productId },
        data: {
          stock: {
            increment: quantity,
          },
        },
      });

      this.logger.log(
        `Updated product stock for productId: ${productId}, incremented by: ${quantity}`,
      );

      this.client.emit('stock_update', { productId, quantity });

      return updatedProduct;
    } catch (error) {
      this.logger.error(
        `Failed to update product stock for productId: ${productId}, error: ${error.message}`,
      );
      throw new InternalServerErrorException(error.message);
    }
  }

}

email.module.ts

import { Module } from '@nestjs/common';
import { RmqModule } from '@app/common/rabbit-mq/rabbitmq.module';
import { EmailService } from './email.service';
import { EmailController } from './email.controller';

@Module({
  imports: [
    RmqModule.register({
      name: 'email',
    }),
  ],
  controllers: [EmailController],
  providers: [EmailService],
  exports: [EmailService],
})
export class EmailModule {}

email.service.ts

import { Injectable } from '@nestjs/common';
import * as nodemailer from 'nodemailer';

@Injectable()
export class EmailService {
  private transporter;

  constructor() {
    this.transporter = nodemailer.createTransport({
      host: process.env.EMAIL_HOST,
      port: Number(process.env.EMAIL_PORT),
      secure: true,
      auth: {
        user: process.env.EMAIL_USER,
        pass: process.env.EMAIL_PASS,
      },
    });
  }

  async sendStockUpdateEmail(productId: string, quantity: number) {
    const info = await this.transporter.sendMail({
      from: '[email protected]',
      to: '[email protected]',
      subject: 'Stock Update Notification',
      text: `The stock for product ${productId} has been updated by ${quantity}.`,
      html: `<b>The stock for product ${productId} has been updated by ${quantity}.</b>`,
    });

    console.log('Message sent: %s', info.messageId);
  }
}

email.controller.ts

import { Controller } from '@nestjs/common';
import { EventPattern, Payload, Ctx, RmqContext } from '@nestjs/microservices';
import { EmailService } from './email.service';

@Controller()
export class EmailController {
  constructor(private readonly emailService: EmailService) {}

  @EventPattern('stock_update')
  async handleStockUpdate(@Payload() data: any, @Ctx() context: RmqContext) {
    const channel = context.getChannelRef();
    const originalMessage = context.getMessage();

    try {
      const { productId, quantity } = data;
      await this.emailService.sendStockUpdateEmail(productId, quantity);
      channel.ack(originalMessage); 
    } catch (error) {
      console.error('Error processing message:', error);
      channel.nack(originalMessage); 
    }
  }
}

logger.module.ts

import { Module } from '@nestjs/common';
import { RmqModule } from '@app/common/rabbit-mq/rabbitmq.module';
import { LogService } from './log.service';
import { LogController } from './log.controller';

@Module({
  imports: [
    RmqModule.register({
      name: 'logger', 
    }),
  ],
  controllers: [LogController],
  providers: [LogService],
})
export class LogModule {}

logger.service.ts

import { Injectable, Logger } from '@nestjs/common';

@Injectable()
export class LogService {
  private readonly logger = new Logger(LogService.name);

  logStockUpdate(productId: string, quantity: number) {
    this.logger.log(
      `Log service: Updated product stock for productId: ${productId}, incremented by: ${quantity}`,
    );
  }
}

logger.controller.ts

import { Controller } from '@nestjs/common';
import { EventPattern, Payload, Ctx, RmqContext } from '@nestjs/microservices';
import { LogService } from './log.service';

@Controller()
export class LogController {
  constructor(private readonly logService: LogService) {}

  @EventPattern('stock_update')
  async handleStockUpdate(@Payload() data: any, @Ctx() context: RmqContext) {
    const channel = context.getChannelRef();
    const originalMessage = context.getMessage();

    try {
      const { productId, quantity } = data;
      await this.logService.logStockUpdate(productId, quantity);
      channel.ack(originalMessage); 
    } catch (error) {
      console.error('Error processing message:', error);
      channel.nack(originalMessage); 
    }
  }
}

main.ts

import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';
import { ValidationPipe } from '@nestjs/common';
import { Logger } from 'nestjs-pino';
import { ConfigService } from '@nestjs/config';
import * as passport from 'passport';
import * as cookieParser from 'cookie-parser';
import { RabbitMQService } from '@app/common/rabbit-mq/rabbitmq.service';

async function bootstrap() {
  const app = await NestFactory.create(AppModule);
  const rmqService = app.get<RabbitMQService>(RabbitMQService);
  app.connectMicroservice(rmqService.getOptions('inventory'));
  app.connectMicroservice(rmqService.getOptions('logger'));
  app.connectMicroservice(rmqService.getOptions('email'));
  await app.startAllMicroservices();
  app.use(cookieParser());
  app.use(passport.initialize());
  app.useGlobalPipes(
    new ValidationPipe({
      whitelist: true,
      transform: true,
      transformOptions: { enableImplicitConversion: true },
    }),
  );
  app.useLogger(app.get(Logger));
  const configService = app.get(ConfigService);
  const port = configService.get('PORT');
  await app.listen(port);
}
bootstrap();

single queue works nice but I dont want that