Microservice Transports

import { NestFactory } from '@nestjs/core';
import type { MicroserviceOptions } from '@nestjs/microservices';
import { ReviewsModule } from '../nats/reviews.module.js';
import { natsOptions } from './options.js';

export async function bootstrap() {
  // A standalone microservice: no HTTP server, just the transporter.
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(
    ReviewsModule,
    natsOptions,
  );
  app.enableShutdownHooks();
  await app.listen();
}

import { NestFactory } from '@nestjs/core';
import { ConfigService } from '@nestjs/config';
import { Transport } from '@nestjs/microservices';
import type { AsyncMicroserviceOptions } from '@nestjs/microservices';
import { ReviewsModule } from '../nats/reviews.module.js';

export async function bootstrapFromConfig() {
  // Transport options that need injected providers: AsyncMicroserviceOptions.
  const app = await NestFactory.createMicroservice<AsyncMicroserviceOptions>(
    ReviewsModule,
    {
      inject: [ConfigService],
      useFactory: (config: ConfigService) => ({
        transport: Transport.NATS,
        options: {
          servers: [config.getOrThrow<string>('NATS_URL')],
          queue: 'notifications',
        },
      }),
    },
  );
  await app.listen();
}

import { join } from 'node:path';
import { Module } from '@nestjs/common';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { ClientsModule, Transport } from '@nestjs/microservices';

@Module({
  imports: [
    // registerAsync() factories live in ClientsModule's own scope: either make
    // ConfigModule global (as here) or add `imports: [ConfigModule]` to each entry.
    ConfigModule.forRoot({ isGlobal: true }),
    ClientsModule.registerAsync([
      {
        name: 'REDIS_CLIENT',
        inject: [ConfigService],
        useFactory: (config: ConfigService) => ({
          transport: Transport.REDIS,
          options: {
            host: config.getOrThrow<string>('REDIS_HOST'),
            port: config.getOrThrow<number>('REDIS_PORT'),
          },
        }),
      },
      {
        name: 'MQTT_CLIENT',
        inject: [ConfigService],
        useFactory: (config: ConfigService) => ({
          transport: Transport.MQTT,
          options: { url: config.getOrThrow<string>('MQTT_URL') },
        }),
      },
      {
        name: 'NATS_CLIENT',
        inject: [ConfigService],
        useFactory: (config: ConfigService) => ({
          transport: Transport.NATS,
          options: { servers: [config.getOrThrow<string>('NATS_URL')] },
        }),
      },
      {
        name: 'RMQ_CLIENT',
        inject: [ConfigService],
        useFactory: (config: ConfigService) => ({
          transport: Transport.RMQ,
          options: {
            urls: [config.getOrThrow<string>('RABBITMQ_URL')],
            queue: 'notifications_queue',
            queueOptions: { durable: true },
          },
        }),
      },
      {
        name: 'KAFKA_CLIENT',
        inject: [ConfigService],
        useFactory: (config: ConfigService) => ({
          transport: Transport.KAFKA,
          options: {
            client: {
              clientId: 'bookshelf',
              brokers: config.getOrThrow<string>('KAFKA_BROKERS').split(','),
            },
            consumer: { groupId: 'bookshelf-consumer' },
          },
        }),
      },
      {
        name: 'GRPC_CLIENT',
        inject: [ConfigService],
        useFactory: (config: ConfigService) => ({
          transport: Transport.GRPC,
          options: {
            package: 'bookshelf',
            protoPath: join(import.meta.dirname, '../grpc/books.proto'),
            url: config.getOrThrow<string>('BOOKS_GRPC_URL'),
            loader: { keepCase: true },
          },
        }),
      },
    ]),
  ],
  exports: [ClientsModule],
})
export class BookshelfClientsModule {}

import { Inject, Injectable } from '@nestjs/common';
import { ClientProxy } from '@nestjs/microservices';
import { firstValueFrom, timeout } from 'rxjs';

@Injectable()
export class ReviewNotifier {
  constructor(@Inject('NATS_CLIENT') private readonly client: ClientProxy) {}

  // Request-response: always bound the wait with timeout().
  countUnread(userId: number): Promise<number> {
    return firstValueFrom(
      this.client.send<number, number>('notifications.count', userId).pipe(timeout(5000)),
    );
  }

  // Event: fire-and-forget. emit() is cold until subscribed, so await it
  // (or subscribe) to make sure the event is actually dispatched.
  notifyReviewCreated(bookId: number): Promise<void> {
    return firstValueFrom(this.client.emit('reviews.created', { bookId }));
  }
}

import { Inject, Injectable } from '@nestjs/common';
import { ClientProxy, RedisEvents, RedisStatus } from '@nestjs/microservices';

@Injectable()
export class RedisConnectionWatcher {
  // Typing the client with the transport's events and status maps makes
  // status and on() type-check under `strict`.
  constructor(
    @Inject('REDIS_CLIENT')
    private readonly client: ClientProxy<RedisEvents, RedisStatus>,
  ) {}

  watch() {
    this.client.status.subscribe((status) => console.log('redis', status));
    // Redis uses two connections, so the callback also receives which one failed.
    this.client.on('error', (connection, err) =>
      console.error(connection, err),
    );
    const [pub, sub] =
      this.client.unwrap<[import('ioredis').Redis, import('ioredis').Redis]>();
    return { pub, sub };
  }
}

import { RedisEvents } from '@nestjs/microservices';
import type { INestMicroservice } from '@nestjs/common';

export function watchServer(app: INestMicroservice) {
  // INestMicroservice.status is Observable<string>: do not annotate the
  // parameter with the transport's status enum.
  app.status.subscribe((status) => console.log('server', status));
  // With only the first type argument the callback parameters are not inferred.
  app.on<RedisEvents, 'error'>('error', (connection, err) =>
    console.error(connection, err),
  );
  const [pub, sub] =
    app.unwrap<[import('ioredis').Redis, import('ioredis').Redis]>();
  return { pub, sub };
}

import { Inject, Injectable } from '@nestjs/common';
import {
  ClientProxy,
  MqttEvents,
  MqttStatus,
  NatsEvents,
  NatsStatus,
  RmqEvents,
  RmqStatus,
} from '@nestjs/microservices';
import type { ClientKafkaProxy } from '@nestjs/microservices';

@Injectable()
export class DriverAccess {
  constructor(
    @Inject('MQTT_CLIENT') private readonly mqtt: ClientProxy<MqttEvents, MqttStatus>,
    @Inject('NATS_CLIENT') private readonly nats: ClientProxy<NatsEvents, NatsStatus>,
    @Inject('RMQ_CLIENT') private readonly rmq: ClientProxy<RmqEvents, RmqStatus>,
    @Inject('KAFKA_CLIENT') private readonly kafka: ClientKafkaProxy,
  ) {}

  inspect() {
    this.mqtt.on('error', (err) => console.error(err));
    const mqttClient = this.mqtt.unwrap<import('mqtt').MqttClient>();

    this.nats.on('disconnect', (serverUrl) => console.error(`Disconnected from ${serverUrl}`));
    const natsConnection =
      this.nats.unwrap<import('@nats-io/transport-node').NatsConnection>();

    this.rmq.on('error', (err) => console.error(err));
    const manager =
      this.rmq.unwrap<import('amqp-connection-manager').AmqpConnectionManager>();

    // Kafka has no unwrap()/on(): the KafkaJS producer and consumer are properties.
    const { producer, consumer } = this.kafka;

    return { mqttClient, natsConnection, manager, producer, consumer };
  }
}

import { MicroserviceOptions, Transport } from '@nestjs/microservices';

export const redisOptions: MicroserviceOptions = {
  transport: Transport.REDIS,
  options: {
    host: 'localhost',
    port: 6379,
    retryAttempts: 5,
    retryDelay: 1000,
    wildcards: true,
  },
};

import { Controller } from '@nestjs/common';
import {
  Ctx,
  EventPattern,
  MessagePattern,
  Payload,
  RedisContext,
} from '@nestjs/microservices';

@Controller()
export class NotificationsController {
  readonly received: { channel: string; data: unknown }[] = [];

  // Request-response: the reply travels on "notifications.count.reply".
  @MessagePattern('notifications.count')
  count(@Payload() userId: number, @Ctx() ctx: RedisContext) {
    return { userId, unread: 3, channel: ctx.getChannel() };
  }

  // With wildcards: true the pattern is subscribed with PSUBSCRIBE.
  @EventPattern<string>('notify.*')
  onNotification(@Payload() data: unknown, @Ctx() ctx: RedisContext) {
    this.received.push({ channel: ctx.getChannel(), data });
  }
}

@Controller()
class Overlap {
  seen: string[] = [];

  @MessagePattern('notifications.count')
  count() {
    return 1;
  }

  @EventPattern<string>('notifications.*')
  any(@Ctx() ctx: RedisContext) {
    this.seen.push(ctx.getChannel());
  }
}
// seen contains 'notifications.count' and 'notifications.count.reply'

import { Controller } from '@nestjs/common';
import {
  Ctx,
  EventPattern,
  MessagePattern,
  MqttContext,
  Payload,
} from '@nestjs/microservices';

@Controller()
export class ReadingsController {
  readonly seen: { topic: string; qos: number; data: unknown }[] = [];

  // '+' matches exactly one topic level, '#' matches the rest of the topic.
  @MessagePattern('shelves/+/temperature/+')
  temperature(@Payload() data: { celsius: number }, @Ctx() ctx: MqttContext) {
    return { topic: ctx.getTopic(), celsius: data.celsius };
  }

  // Per-pattern QoS overrides the global subscribeOptions.qos.
  @EventPattern<string>('books/+/reviews', { qos: 1 })
  review(@Payload() data: { text: string }, @Ctx() ctx: MqttContext) {
    this.seen.push({
      topic: ctx.getTopic(),
      qos: ctx.getPacket().qos ?? 0,
      data,
    });
  }

  @EventPattern<string>('audit/#')
  audit(@Payload() data: unknown, @Ctx() ctx: MqttContext) {
    this.seen.push({ topic: ctx.getTopic(), qos: ctx.getPacket().qos ?? 0, data });
  }
}

import { ClientProxyFactory, MqttRecordBuilder, Transport } from '@nestjs/microservices';

// MQTT 5 user properties (the in-process broker used for the runnable tests is MQTT 3.1.1 only).
export const mqttRecord = new MqttRecordBuilder('payload')
  .setProperties({ userProperties: { 'x-version': '1.0.0' } })
  .setQoS(1)
  .build();

export const mqttV5Client = ClientProxyFactory.create({
  transport: Transport.MQTT,
  options: {
    url: 'mqtt://localhost:1883',
    protocolVersion: 5,
    userProperties: { 'x-version': '1.0.0' },
    subscribeOptions: { qos: 1 },
  },
});

import { MicroserviceOptions, Transport } from '@nestjs/microservices';

export const natsOptions: MicroserviceOptions = {
  transport: Transport.NATS,
  options: {
    servers: ['nats://localhost:4222'],
    queue: 'reviews-workers',
    gracefulShutdown: true,
    gracePeriod: 10000,
    headers: { 'x-version': '1.0.0' },
  },
};

import { Controller } from '@nestjs/common';
import {
  Ctx,
  EventPattern,
  MessagePattern,
  NatsContext,
  Payload,
} from '@nestjs/microservices';

@Controller()
export class ReviewsController {
  readonly handled: { subject: string; data: unknown }[] = [];

  // Subject wildcards: '*' matches one token, '>' matches one or more tokens.
  @MessagePattern('time.us.*')
  time(@Payload() zone: string, @Ctx() ctx: NatsContext) {
    return { subject: ctx.getSubject(), zone };
  }

  @MessagePattern('replace-emoji')
  replaceEmoji(@Payload() text: string, @Ctx() ctx: NatsContext) {
    const headers = ctx.getHeaders();
    return headers?.get('x-version') === '1.0.0' ? '🐱' : '🐈';
  }

  @EventPattern<string>('reviews.created')
  created(@Payload() data: unknown, @Ctx() ctx: NatsContext) {
    this.handled.push({ subject: ctx.getSubject(), data });
  }
}

const app = await NestFactory.createMicroservice<MicroserviceOptions>(ReviewsModule, {
  transport: Transport.NATS,
  options: {
    servers: ['nats://localhost:4222'],
    queue: 'reviews-workers',
  },
});

import { headers } from '@nats-io/nats-core';
import { NatsRecordBuilder } from '@nestjs/microservices';

const h = headers();
h.set('x-version', '1.0.0');
const record = new NatsRecordBuilder(':cat:').setHeaders(h).build();
this.client.send('replace-emoji', record).subscribe();

import { MicroserviceOptions, Transport } from '@nestjs/microservices';

export const rmqOptions: MicroserviceOptions = {
  transport: Transport.RMQ,
  options: {
    urls: ['amqp://localhost:5672'],
    queue: 'bookshelf_queue',
    prefetchCount: 10,
    isGlobalPrefetchCount: false,
    noAck: false,
    queueOptions: { durable: true },
    persistent: true,
    exchange: 'bookshelf',
    exchangeType: 'topic',
    wildcards: true,
    routingKey: 'reviews.#',
    maxConnectionAttempts: 5,
  },
};

import { Controller } from '@nestjs/common';
import {
  Ctx,
  EventPattern,
  MessagePattern,
  Payload,
  RmqContext,
} from '@nestjs/microservices';

@Controller()
export class RmqNotificationsController {
  readonly handled: { pattern: string; priority?: number; version?: string }[] = [];

  // noAck: false -> the handler MUST acknowledge, otherwise RabbitMQ keeps the
  // message unacknowledged (and redelivers it when the channel closes).
  @MessagePattern('notifications.render')
  render(@Payload() data: { user: string }, @Ctx() context: RmqContext) {
    const channel = context.getChannelRef();
    const original = context.getMessage();
    const { headers, priority } = original.properties;
    this.handled.push({
      pattern: context.getPattern(),
      priority,
      version: headers?.['x-version'],
    });
    channel.ack(original);
    return `Hello ${data.user}`;
  }

  // Events are acknowledged the same way.
  @EventPattern<string>('notifications.sent')
  sent(@Payload() data: unknown, @Ctx() context: RmqContext) {
    this.handled.push({ pattern: context.getPattern() });
    context.getChannelRef().ack(context.getMessage());
  }
}

import { RmqRecordBuilder } from '@nestjs/microservices';

const record = new RmqRecordBuilder({ user: 'ana' })
  .setOptions({
    headers: { 'x-version': '1.0.0' },
    priority: 3,
  })
  .build();

this.client.send('notifications.render', record).subscribe();

import { Controller } from '@nestjs/common';
import {
  Ctx,
  EventPattern,
  MessagePattern,
  Payload,
  RmqContext,
} from '@nestjs/microservices';

@Controller()
export class RmqTopicController {
  readonly routingKeys: string[] = [];

  // '#' matches zero or more words, '*' exactly one word.
  @EventPattern<string>('reviews.#')
  onReviews(@Payload() data: unknown, @Ctx() context: RmqContext) {
    this.routingKeys.push(context.getPattern());
  }

  @MessagePattern('books.*')
  book(@Payload() data: { id: number }, @Ctx() context: RmqContext) {
    return { routingKey: context.getPattern(), id: data.id };
  }
}

import { MicroserviceOptions, Transport } from '@nestjs/microservices';

export const kafkaOptions: MicroserviceOptions = {
  transport: Transport.KAFKA,
  options: {
    client: { clientId: 'bookshelf', brokers: ['localhost:9092'] },
    consumer: { groupId: 'bookshelf-consumer' },
    subscribe: { fromBeginning: true },
    run: { autoCommit: false },
    send: { acks: -1 },
    postfixId: '-server',
  },
};

import { Inject, Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
import type { ClientKafkaProxy } from '@nestjs/microservices';
import { firstValueFrom, timeout } from 'rxjs';

@Injectable()
export class BooksKafkaGateway implements OnModuleInit, OnModuleDestroy {
  constructor(@Inject('KAFKA_CLIENT') private readonly client: ClientKafkaProxy) {}

  async onModuleInit() {
    // Required for request-response: subscribe to "books.get.reply" BEFORE connecting.
    this.client.subscribeToResponseOf('books.get');
    await this.client.connect();
  }

  async onModuleDestroy() {
    await this.client.close();
  }

  getBook(id: number) {
    return firstValueFrom(
      this.client.send<{ id: number; title: string }, { id: number }>('books.get', { id }).pipe(timeout(5000)),
    );
  }

  // Events need no reply topic.
  reviewCreated(bookId: number) {
    return firstValueFrom(this.client.emit('reviews.created', { bookId }));
  }
}

import { Controller } from '@nestjs/common';
import {
  Ctx,
  EventPattern,
  KafkaContext,
  KafkaRetriableException,
  MessagePattern,
  Payload,
} from '@nestjs/microservices';

@Controller()
export class BooksKafkaController {
  readonly log: string[] = [];
  private flakyCalls = 0;

  // Request-response: the answer goes to "books.get.reply". Returning
  // { key, value, headers } keys and annotates the reply message.
  @MessagePattern('books.get')
  get(@Payload() message: { id: number }) {
    return {
      key: String(message.id),
      value: { id: message.id, title: 'Dune' },
      headers: { kafka_nestRealm: 'bookshelf' },
    };
  }

  // A KafkaRetriableException is not turned into an error reply: it bubbles up
  // to KafkaJS, which redelivers the message (the offset is not committed).
  @MessagePattern('books.flaky')
  flaky(@Payload() message: { id: number }) {
    this.flakyCalls += 1;
    if (this.flakyCalls < 3) throw new KafkaRetriableException('try again');
    return { id: message.id, attempts: this.flakyCalls };
  }

  // A RegExp pattern subscribes to every matching topic (Nest 12).
  @EventPattern(/^reviews\..+$/)
  onReview(@Payload() data: unknown, @Ctx() context: KafkaContext) {
    this.log.push(`${context.getTopic()}#${context.getPartition()}`);
  }
}

import { Controller } from '@nestjs/common';
import {
  Ctx,
  EventPattern,
  KafkaContext,
  Payload,
} from '@nestjs/microservices';

@Controller()
export class ManualCommitController {
  readonly processed: unknown[] = [];

  // Pair with `run: { autoCommit: false }` in the server options.
  @EventPattern<string>('payments.settled')
  async settled(@Payload() data: unknown, @Ctx() context: KafkaContext) {
    this.processed.push(data); // business logic first ...

    const { offset } = context.getMessage();
    const consumer = context.getConsumer();
    // ... then commit the NEXT offset to consume (current offset + 1).
    await consumer.commitOffsets([
      {
        topic: context.getTopic(),
        partition: context.getPartition(),
        offset: (Number(offset) + 1).toString(),
      },
    ]);
  }
}

import { ArgumentsHost, Catch, ExceptionFilter, Logger } from '@nestjs/common';
import { KafkaContext } from '@nestjs/microservices';
import type { Producer } from 'kafkajs';

const toWire = (v: unknown) =>
  v === null || typeof v === 'string' || Buffer.isBuffer(v)
    ? v
    : JSON.stringify(v);

/**
 * Retries a failing event a bounded number of times by republishing it with a
 * retry-count header, then skips it (committing its offset) so the partition
 * is not blocked by a poison message.
 */
@Catch()
export class KafkaMaxRetryExceptionFilter implements ExceptionFilter {
  private readonly logger = new Logger(KafkaMaxRetryExceptionFilter.name);

  constructor(
    private readonly producer: Producer,
    private readonly maxRetries: number,
    private readonly skipHandler?: (message: unknown) => Promise<void>,
  ) {}

  async catch(exception: unknown, host: ArgumentsHost) {
    const context = host.switchToRpc().getContext<KafkaContext>();
    const message = context.getMessage();
    const retries = this.retryCount(context);

    if (retries >= this.maxRetries) {
      this.logger.warn(`Max retries (${this.maxRetries}) exceeded, skipping`);
      await this.skipHandler?.(message);
      await this.commit(context);
      return; // stop propagating: KafkaJS sees a successful handler
    }

    try {
      await this.producer.send({
        topic: context.getTopic(),
        messages: [
          {
            key: message.key,
            // Nest has already JSON-parsed the value; KafkaJS only accepts
            // strings and Buffers, so serialize it again.
            value: toWire(message.value),
            headers: { ...message.headers, 'retry-count': String(retries + 1) },
          },
        ],
      });
      await this.commit(context);
    } catch (republishError) {
      this.logger.error('Failed to republish for retry', republishError);
      throw exception; // let KafkaJS redeliver the original message
    }
  }

  private retryCount(context: KafkaContext): number {
    const header = context.getMessage().headers?.['retry-count'];
    return header ? parseInt(String(header), 10) || 0 : 0;
  }

  private async commit(context: KafkaContext) {
    await context.getConsumer().commitOffsets([
      {
        topic: context.getTopic(),
        partition: context.getPartition(),
        // commit the NEXT offset: current offset + 1
        offset: (Number(context.getMessage().offset) + 1).toString(),
      },
    ]);
  }
}

import { Controller, Inject, Injectable, UseFilters } from '@nestjs/common';
import { Payload, EventPattern } from '@nestjs/microservices';
import type { Producer } from 'kafkajs';
import { KafkaMaxRetryExceptionFilter } from './kafka-max-retry.filter.js';

@Injectable()
export class AppKafkaRetryFilter extends KafkaMaxRetryExceptionFilter {
  constructor(@Inject('KAFKA_PRODUCER') producer: Producer) {
    super(producer, 2); // maxRetries = 2
  }
}

@Controller()
@UseFilters(AppKafkaRetryFilter)
export class RetryController {
  readonly attempts: number[] = [];

  @EventPattern<string>('emails.send')
  async send(@Payload() data: { n: number }) {
    this.attempts.push(data.n);
    throw new Error('SMTP down'); // always fails -> retried twice, then skipped
  }
}

import { Module } from '@nestjs/common';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { Kafka } from 'kafkajs';
import { AppKafkaRetryFilter, RetryController } from './retry.controller.js';
import { BooksKafkaController } from './books.controller.js';
import { ManualCommitController } from './manual-commit.controller.js';

@Module({
  imports: [ConfigModule.forRoot()],
  controllers: [BooksKafkaController, ManualCommitController, RetryController],
  providers: [
    AppKafkaRetryFilter,
    {
      provide: 'KAFKA_PRODUCER',
      inject: [ConfigService],
      useFactory: async (config: ConfigService) => {
        const kafka = new Kafka({
          brokers: config.getOrThrow<string>('KAFKA_BROKERS').split(','),
        });
        const producer = kafka.producer();
        await producer.connect();
        return producer;
      },
    },
  ],
})
export class KafkaAppModule {}

import { readFileSync } from 'node:fs';
import { join } from 'node:path';
import { ServerCredentials } from '@grpc/grpc-js';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';

// gRPC over TLS: certificate paths come from configuration, never from literals.
export const grpcTls = (dir: string): MicroserviceOptions => ({
  transport: Transport.GRPC,
  options: {
    package: 'bookshelf',
    protoPath: join(dir, 'books.proto'),
    url: '0.0.0.0:50051',
    credentials: ServerCredentials.createSsl(
      readFileSync(join(dir, 'ca.pem')),
      [{ cert_chain: readFileSync(join(dir, 'server.pem')), private_key: readFileSync(join(dir, 'server.key')) }],
      false,
    ),
  },
});

syntax = "proto3";

package bookshelf;

service BooksService {
  // unary
  rpc FindOne (BookById) returns (Book) {}
  rpc Create (Book) returns (Book) {}
  // server streaming
  rpc ListReviews (BookById) returns (stream Review) {}
  // client streaming
  rpc RateMany (stream Rating) returns (RatingSummary) {}
  // bidirectional streaming
  rpc Chat (stream ChatMessage) returns (stream ChatMessage) {}
  // client streaming handled with @GrpcStreamCall (callback style)
  rpc CountRatings (stream Rating) returns (RatingSummary) {}
  // bidirectional streaming handled with @GrpcStreamCall (duplex stream)
  rpc Echo (stream ChatMessage) returns (stream ChatMessage) {}
}

message BookById {
  int32 id = 1;
}

message Book {
  int32 id = 1;
  string title = 2;
}

message Review {
  int32 book_id = 1;
  string text = 2;
}

message Rating {
  int32 book_id = 1;
  int32 stars = 2;
}

message RatingSummary {
  int32 count = 1;
  double average = 2;
}

message ChatMessage {
  string text = 1;
}

import { Controller } from '@nestjs/common';
import {
  GrpcAlreadyExistsException,
  GrpcMethod,
  GrpcNotFoundException,
  GrpcStreamCall,
  GrpcStreamMethod,
} from '@nestjs/microservices';
import { Metadata } from '@grpc/grpc-js';
import type { ServerDuplexStream, ServerUnaryCall } from '@grpc/grpc-js';
import { Observable, Subject, from, map, reduce } from 'rxjs';

export interface Book {
  id: number;
  title: string;
}
export interface BookById {
  id: number;
}
export interface Review {
  book_id: number;
  text: string;
}
export interface Rating {
  book_id: number;
  stars: number;
}
export interface RatingSummary {
  count: number;
  average: number;
}
export interface ChatMessage {
  text: string;
}

@Controller()
export class BooksController {
  private readonly books = new Map<number, Book>([
    [1, { id: 1, title: 'Dune' }],
    [2, { id: 2, title: 'Emma' }],
  ]);

  // Both arguments given: service name and rpc name.
  @GrpcMethod('BooksService', 'FindOne')
  findOne(
    data: BookById,
    metadata: Metadata,
    call: ServerUnaryCall<BookById, Book>,
  ): Book {
    // Echo the request id back as initial (header) metadata.
    const out = new Metadata();
    out.add('x-request-id', String(metadata.get('x-request-id')[0] ?? 'none'));
    call.sendMetadata(out);

    const book = this.books.get(data.id);
    if (!book) throw new GrpcNotFoundException(`Book ${data.id} not found`);
    return book;
  }

  // Only the service name given: the rpc is derived from the method name
  // (create -> Create).
  @GrpcMethod('BooksService')
  create(data: Book): Book {
    if (this.books.has(data.id)) {
      throw new GrpcAlreadyExistsException('Book already exists');
    }
    this.books.set(data.id, data);
    return data;
  }

  // Server streaming: return an Observable, every value is a stream message.
  @GrpcMethod('BooksService', 'ListReviews')
  listReviews(data: BookById): Observable<Review> {
    return from([
      { book_id: data.id, text: 'Great' },
      { book_id: data.id, text: 'Long' },
      { book_id: data.id, text: 'Classic' },
    ]);
  }

  // Client streaming with the RxJS strategy: the last value emitted is the
  // single response.
  @GrpcStreamMethod('BooksService', 'RateMany')
  rateMany(ratings: Observable<Rating>): Observable<RatingSummary> {
    return ratings.pipe(
      reduce(
        (acc, r) => ({ count: acc.count + 1, sum: acc.sum + r.stars }),
        { count: 0, sum: 0 },
      ),
      map(({ count, sum }) => ({
        count,
        average: count === 0 ? 0 : sum / count,
      })),
    );
  }

  // Bidirectional streaming with the Subject strategy.
  @GrpcStreamMethod('BooksService', 'Chat')
  chat(messages: Observable<ChatMessage>): Observable<ChatMessage> {
    const subject = new Subject<ChatMessage>();
    messages.subscribe({
      next: (m) => subject.next({ text: `echo: ${m.text}` }),
      complete: () => subject.complete(),
    });
    return subject.asObservable();
  }

  // Client streaming with the call-stream handler: respond through the callback.
  @GrpcStreamCall('BooksService', 'CountRatings')
  countRatings(
    requestStream: ServerDuplexStream<Rating, RatingSummary>,
    callback: (err: unknown, value: RatingSummary) => void,
  ) {
    let count = 0;
    let sum = 0;
    requestStream.on('data', (r: Rating) => {
      count += 1;
      sum += r.stars;
    });
    requestStream.on('end', () =>
      callback(null, { count, average: count === 0 ? 0 : sum / count }),
    );
  }

  // Bidirectional streaming with the call-stream handler: a plain Duplex.
  @GrpcStreamCall('BooksService', 'Echo')
  echo(call: ServerDuplexStream<ChatMessage, ChatMessage>) {
    call.on('data', (m: ChatMessage) => call.write({ text: `echo: ${m.text}` }));
    call.on('end', () => call.end());
  }
}

import { join } from 'node:path';
import { NestFactory } from '@nestjs/core';
import {
  GrpcExceptionFilter,
  MicroserviceOptions,
  Transport,
} from '@nestjs/microservices';
import { ReflectionService } from '@grpc/reflection';
import {
  HealthImplementation,
  protoPath as healthProtoPath,
} from 'grpc-health-check';
import { BooksModule } from './books.module.js';

export async function startBooksGrpc(url = '0.0.0.0:50151') {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(
    BooksModule,
    {
      transport: Transport.GRPC,
      options: {
        package: 'bookshelf',
        protoPath: [healthProtoPath, join(import.meta.dirname, 'books.proto')],
        url,
        loader: { keepCase: true },
        onLoadPackageDefinition: (pkg, server) => {
          new ReflectionService(pkg).addToServer(server);
          const health = new HealthImplementation({ '': 'UNKNOWN' });
          health.addToServer(server);
          health.setStatus('', 'SERVING');
        },
      },
    },
  );
  app.useGlobalFilters(new GrpcExceptionFilter());
  await app.listen();
  return app;
}

import { Inject, Injectable, Module, OnModuleInit } from '@nestjs/common';
import { ClientsModule, Transport } from '@nestjs/microservices';
import type { ClientGrpc } from '@nestjs/microservices';
import { join } from 'node:path';
import { Metadata } from '@grpc/grpc-js';
import { Observable, ReplaySubject, lastValueFrom, toArray } from 'rxjs';
import type {
  Book,
  BookById,
  ChatMessage,
  Rating,
  RatingSummary,
  Review,
} from './books.controller.js';

interface BooksService {
  findOne(data: BookById, metadata?: Metadata): Observable<Book>;
  create(data: Book): Observable<Book>;
  listReviews(data: BookById): Observable<Review>;
  rateMany(upstream: Observable<Rating>): Observable<RatingSummary>;
  chat(upstream: Observable<ChatMessage>): Observable<ChatMessage>;
}

@Injectable()
export class BooksClient implements OnModuleInit {
  private books!: BooksService;

  constructor(@Inject('BOOKS_PACKAGE') private readonly client: ClientGrpc) {}

  onModuleInit() {
    this.books = this.client.getService<BooksService>('BooksService');
  }

  findOne(id: number) {
    const metadata = new Metadata();
    metadata.add('x-request-id', 'req-42');
    return lastValueFrom(this.books.findOne({ id }, metadata));
  }

  create(book: Book) {
    return lastValueFrom(this.books.create(book));
  }

  reviews(id: number) {
    return lastValueFrom(this.books.listReviews({ id }).pipe(toArray()));
  }

  rate(stars: number[]) {
    const upstream = new ReplaySubject<Rating>();
    stars.forEach((s) => upstream.next({ book_id: 1, stars: s }));
    upstream.complete();
    return lastValueFrom(this.books.rateMany(upstream));
  }

  chat(texts: string[]) {
    const upstream = new ReplaySubject<ChatMessage>();
    texts.forEach((text) => upstream.next({ text }));
    upstream.complete();
    return lastValueFrom(this.books.chat(upstream).pipe(toArray()));
  }
}

@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'BOOKS_PACKAGE',
        transport: Transport.GRPC,
        options: {
          package: 'bookshelf',
          protoPath: join(import.meta.dirname, 'books.proto'),
          url: 'localhost:50151',
          loader: { keepCase: true },
        },
      },
    ]),
  ],
  providers: [BooksClient],
  exports: [BooksClient],
})
export class BooksClientModule {}

import { GrpcException, GrpcStatus } from '@nestjs/microservices';

throw new GrpcException('Rate limit exceeded', GrpcStatus.RESOURCE_EXHAUSTED);

requestStream.on('metadata', (metadata: Metadata) => {
  const meta = metadata.get('x-meta');
});

npm install @grpc/reflection grpc-health-check