Transports

A transporter is the driver that carries a packet between Nest processes; your handlers stay the same whichever one you pick.

SyntaxNestFactory.createMicroservice<T extends object>(moduleCls, options?): Promise<INestMicroservice>

A Nest microservice is an application that listens on something other than HTTP. There is no URL and no verb. An incoming packet carries a pattern - a string or a plain object that says which handler should run - plus a payload. A transporter is the driver that moves that packet: it owns the socket or broker connection, serialises outgoing packets, and hands incoming ones to Nest's dispatcher.

One model, seven drivers

Handlers never learn which transporter is underneath. You change the driver in main.ts and the method body is untouched. Transport is an enum exported by @nestjs/microservices, and its members are upper case.

TransportPackage you installReply channel
TCP (default)nonesame socket
REDISioredischannel <route>.reply
NATS@nats-io/transport-nodeper-request reply subject
RMQamqplib, amqp-connection-manageramq.rabbitmq.reply-to
KAFKAkafkajstopic <topic>.reply

What trips people up

The abstraction stops at delivery guarantees. A Redis channel is fire-and-forget: publish while nobody is subscribed and the message is gone. RabbitMQ and Kafka keep it. Same handler code, different failure mode, so the choice of driver is an architectural decision and not a config detail. Two NestJS 12 specifics: the NATS transporter now targets NATS v3 through @nats-io/transport-node, so uninstall the old nats package when you upgrade, and the Kafka transporter accepts a RegExp pattern directly.

Example

Example · typescript
// src/main.ts
import { ValidationPipe } from '@nestjs/common';
import { NestFactory } from '@nestjs/core';
import { MicroserviceOptions, RpcException, Transport } from '@nestjs/microservices';
import { AppModule } from './app.module.js';

async function bootstrap() {
  // Transport.TCP is the default, so `transport` could be left out entirely.
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
    transport: Transport.TCP,
    options: {
      host: '0.0.0.0',
      port: 3001,
      retryAttempts: 5, // reconnection attempts, not per-message retries
      retryDelay: 1000, // ms between those attempts
    },
  });

  // Global pipes bind to every pattern handler, exactly as they bind to routes.
  // ValidationPipe's default error is an HttpException, and an RPC caller only
  // ever sees { status: 'error', message: 'Internal server error' } for that,
  // so convert it into an RpcException the caller can actually read.
  app.useGlobalPipes(
    new ValidationPipe({
      whitelist: true,
      exceptionFactory: errors =>
        new RpcException({
          status: 'error',
          message: errors.flatMap(e => Object.values(e.constraints ?? {})),
        }),
    }),
  );

  await app.listen(); // resolves once the transporter is accepting messages
}

bootstrap();

When to use it

  • Run billing on a private TCP port that only the API gateway can reach.
  • Fan an order.paid event out over Redis without the publisher knowing who listens.
  • Put a Kafka topic between ingest and its consumers so a slow consumer cannot block ingest.
  • Add a transport listener to an existing HTTP app so internal workers skip the public URL.

More examples

One process, two transporters

A hybrid app is the cheapest way to add a transport: the same controllers serve HTTP and RPC, and you deploy one artefact. Reach for it when an internal caller should not go through your public URL, or as a staging post before a real split. The transport argument on @MessagePattern matters only once an app listens on more than one driver.

Example · typescript
// src/main.ts
import { NestFactory } from '@nestjs/core';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';
import { AppModule } from './app.module.js';

async function bootstrap() {
  const app = await NestFactory.create(AppModule);

  // A hybrid app: HTTP routes plus two transport listeners in one deployable.
  app.connectMicroservice<MicroserviceOptions>({
    transport: Transport.TCP,
    options: { port: 3001 },
  });
  app.connectMicroservice<MicroserviceOptions>({
    transport: Transport.REDIS,
    options: { host: 'localhost', port: 6379 },
  });

  // Required. connectMicroservice() only registers the listeners.
  await app.startAllMicroservices();
  await app.listen(3000);
}

bootstrap();

// src/billing/billing.controller.ts
import { Controller } from '@nestjs/common';
import { Ctx, MessagePattern, Payload, RedisContext, Transport } from '@nestjs/microservices';

@Controller()
export class BillingController {
  // The second argument pins this handler to one transporter. Without it the
  // handler answers on every transporter the app is listening on.
  @MessagePattern({ cmd: 'invoice.total' }, Transport.REDIS)
  invoiceTotal(@Payload() lines: number[], @Ctx() ctx: RedisContext): number {
    // The Redis channel is the normalised pattern: '{"cmd":"invoice.total"}'
    console.log(ctx.getChannel());
    return lines.reduce((sum, cents) => sum + cents, 0);
  }
}

Registering clients: shorthand and longhand

ClientsModule is how the calling side gets a client. The array form and the object form are the same options type; the object form exists so you can pass isGlobal. Use registerAsync whenever a host or broker URL comes from the environment, because the factory runs during bootstrap and can inject ConfigService.

Example · typescript
// src/app.module.ts
import { Module } from '@nestjs/common';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { ClientsModule, Transport } from '@nestjs/microservices';
import { OrdersController } from './orders/orders.controller.js';

@Module({
  imports: [
    // Shorthand: an array of client definitions. Each `name` becomes the
    // injection token you later @Inject() to get a ClientProxy.
    ClientsModule.register([
      { name: 'BILLING', transport: Transport.TCP, options: { port: 3001 } },
    ]),

    // Longhand: the same registration as an object. Identical, except that
    // this form also accepts isGlobal, which exports the clients everywhere
    // so other modules need not import ClientsModule again.
    // ClientsModule.register({
    //   clients: [
    //     { name: 'BILLING', transport: Transport.TCP, options: { port: 3001 } },
    //   ],
    //   isGlobal: true,
    // }),

    // When the address comes from configuration, build it in a factory.
    ClientsModule.registerAsync([
      {
        name: 'SHIPPING',
        imports: [ConfigModule],
        inject: [ConfigService],
        useFactory: (config: ConfigService) => ({
          transport: Transport.NATS,
          options: { servers: [config.getOrThrow<string>('NATS_URL')] },
        }),
      },
    ]),
  ],
  controllers: [OrdersController],
})
export class AppModule {}

Kafka: subscribe to replies before connecting

Kafka is the one transporter where request-response needs setup on the client. You inject ClientKafkaProxy rather than ClientProxy, call subscribeToResponseOf() for each pattern you will send, and only then connect. Skipping it produces a send() that never resolves, which looks exactly like a dead service.

Example · typescript
// src/orders/orders.service.ts
import { Inject, Injectable, OnModuleInit } from '@nestjs/common';
import { ClientKafkaProxy } from '@nestjs/microservices';
import { firstValueFrom, timeout } from 'rxjs';

interface Quote {
  cents: number;
}

@Injectable()
export class OrdersService implements OnModuleInit {
  constructor(@Inject('BILLING') private readonly client: ClientKafkaProxy) {}

  async onModuleInit() {
    // Request-response over Kafka uses a second topic, '<topic>.reply'.
    // The client must declare which replies it wants BEFORE it connects,
    // otherwise the consumer is never subscribed to the reply topic.
    this.client.subscribeToResponseOf('billing.quote');
    await this.client.connect();

    // KafkaStatus: 'connected' | 'disconnected' | 'rebalancing' | 'crashed' | 'stopped'
    this.client.status.subscribe(status => console.log(`kafka: ${status}`));
  }

  quote(orderId: string): Promise<Quote> {
    return firstValueFrom(
      this.client.send<Quote>('billing.quote', { orderId }).pipe(timeout(5000)),
    );
  }
}

Related

nest micro patternsnest micro contractsnest micro resiliencenest adv microservicesnest bootstrap

Discussion

  • Be the first to comment on this lesson.