import { Controller, Get } from '@nestjs/common';
import {
  ClientProxy,
  ClientProxyFactory,
  MessagePattern,
  Transport,
} from '@nestjs/microservices';
import { Observable } from 'rxjs';
import { scan, take } from 'rxjs/operators';

@Controller()
export class RMQBroadcastController {
  client: ClientProxy;

  constructor() {
    this.client = ClientProxyFactory.create({
      transport: Transport.RMQ,
      options: {
        urls: [`amqp://localhost:5672`],
        queue: 'test_broadcast',
        queueOptions: { durable: false },
      },
    });
  }

  @Get('broadcast')
  multicats() {
    return this.client
      .send<number>({ cmd: 'broadcast' }, {})
      .pipe(scan((a, b) => a + b), take(2));
  }

  @MessagePattern({ cmd: 'broadcast' })
  replyBroadcast(): Observable<number> {
    return new Observable(observer => observer.next(1));
  }
}

import { Controller, Get } from '@nestjs/common';
import {
  ClientProxy,
  ClientProxyFactory,
  MessagePattern,
  Transport,
} from '@nestjs/microservices';
import { Observable } from 'rxjs';
import { scan, take } from 'rxjs/operators';

@Controller()
export class RMQBroadcastController {
  client: ClientProxy;

  constructor() {
    this.client = ClientProxyFactory.create({
      transport: Transport.RMQ,
      options: {
        urls: [`amqp://localhost:5672`],
        queue: 'test_broadcast',
        queueOptions: { durable: false },
        socketOptions: { noDelay: true },
      },
    });
  }

  @Get('broadcast')
  multicats() {
    return this.client
      .send<number>({ cmd: 'broadcast' }, {})
      .pipe(scan((a, b) => a + b), take(2));
  }

  @MessagePattern({ cmd: 'broadcast' })
  replyBroadcast(): Observable<number> {
    return new Observable(observer => observer.next(1));
  }
}

Add socketOptions to RMQ broadcast test
Add socketOptions to RMQ broadcast test