Skip to main content

Quick Start

Use whenyou want a working service before reading anything else.

You geta module registration, a bootstrap, one event handler and one RPC handler, running in five minutes.

Five minutes from now you'll have a NestJS service running on NATS JetStream. It emits order.created and answers order.get RPCs, and a restart no longer loses whatever was in flight. Register the module, connect the transport, define handlers, send messages.

Prerequisites

Make sure you have installed the library and have a NATS server running with JetStream enabled.

1. Register the module​

The library uses a root + feature registration pattern:

  • forRoot() initializes the NATS connection and infrastructure (once, in the root module)
  • forFeature() creates a lightweight client for sending messages to a target service
src/app.module.ts
import { Module } from '@nestjs/common';
import { JetstreamModule } from '@horizon-republic/nestjs-jetstream';
import { OrdersController } from './orders.controller';
import { GatewayController } from './gateway.controller';

@Module({
  imports: [
    // Global setup, creates NATS connection, streams, consumers
    JetstreamModule.forRoot({
      name: 'orders',
      servers: ['nats://localhost:4222'],
    }),

    // Client for sending messages to the "orders" service
    JetstreamModule.forFeature({ name: 'orders' }),
  ],
  controllers: [OrdersController, GatewayController],
})
export class AppModule {}
Naming matters

The name in forRoot() identifies this service. The name in forFeature() identifies the target service you want to send messages to. When a service sends messages to itself, the two names are the same.

2. Connect the transport​

In your main.ts, resolve the JetstreamStrategy from the DI container and connect it as a microservice transport:

src/main.ts
import { NestFactory } from '@nestjs/core';
import { JetstreamStrategy } from '@horizon-republic/nestjs-jetstream';
import { AppModule } from './app.module';

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

  // Retrieve the strategy instance from the DI container
  app.connectMicroservice(
    { strategy: app.get(JetstreamStrategy) },
    { inheritAppConfig: true },
  );

  // Required so the JetStream transport drains in-flight handlers on SIGTERM.
  // Skip this and messages in flight when the pod dies will be redelivered
  // only after `ack_wait` expires, slower recovery, noisier metrics.
  app.enableShutdownHooks();

  await app.startAllMicroservices();
  await app.listen(3000);
};

void bootstrap();
Don't instantiate the strategy manually

Unlike other NestJS transports, you must not create the strategy with new JetstreamStrategy(). The module creates it through DI with all required dependencies. Always use app.get(JetstreamStrategy) to retrieve it.

Talking to more than one NATS cluster?

Replace the connectMicroservice call with connectJetstreamMicroservices(app), which attaches one microservice per configured connection. See Multiple connections.

3. Define handlers​

Use standard NestJS decorators to define message handlers:

src/orders.controller.ts
import { Controller, Logger } from '@nestjs/common';
import { Ctx, EventPattern, MessagePattern, Payload } from '@nestjs/microservices';
import { RpcContext } from '@horizon-republic/nestjs-jetstream';

@Controller()
export class OrdersController {
  private readonly logger = new Logger(OrdersController.name);

  /**
   * Workqueue event handler.
   * Each message is delivered to exactly one instance (load-balanced).
   * Acked automatically after successful execution.
   */
  @EventPattern('order.created')
  handleOrderCreated(@Payload() data: { orderId: number; total: number }): void {
    this.logger.log(`Processing order ${data.orderId}, total: $${data.total}`);
    // If this throws, the message is nak'd and redelivered (up to max_deliver, default 3)
  }

  /**
   * Broadcast event handler.
   * Every running instance receives this message.
   */
  @EventPattern('config.updated', { broadcast: true })
  handleConfigUpdated(@Payload() data: { key: string; value: string }): void {
    this.logger.log(`Config changed: ${data.key} = ${data.value}`);
  }

  /**
   * RPC handler.
   * Return value is sent back to the caller.
   */
  @MessagePattern('order.get')
  getOrder(
    @Payload() data: { id: number },
    @Ctx() ctx: RpcContext,
  ): { id: number; status: string } {
    this.logger.log(`RPC: order.get (id=${data.id}), subject: ${ctx.getSubject()}`);
    return { id: data.id, status: 'shipped' };
  }
}

The decorators above cover the handler shapes you'll meet first. A workqueue event goes to one instance, while a broadcast event goes to all of them. An RPC command gets a reply, and strict-order delivery is a fourth shape, reached with @EventPattern('...', { ordered: true }).

For the full picture of each, see Events, Broadcast, Ordered Events, and RPC.

4. Send messages​

Inject the client by the service name you used in forFeature() and use the standard ClientProxy API:

src/gateway.controller.ts
import { Controller, Get, Inject, Param, ParseIntPipe } from '@nestjs/common';
import { ClientProxy } from '@nestjs/microservices';
import { Observable } from 'rxjs';

@Controller('orders')
export class GatewayController {
  constructor(
    @Inject('orders') private readonly client: ClientProxy,
  ) {}

  /** Emit a workqueue event (fire-and-forget, at-least-once delivery). */
  @Get('create')
  createOrder(): Observable<void> {
    return this.client.emit('order.created', { orderId: 42, total: 99.99 });
  }

  /** Emit a broadcast event (all service instances receive it). */
  @Get('broadcast')
  broadcastConfig(): Observable<void> {
    return this.client.emit('broadcast:config.updated', {
      key: 'maintenance',
      value: 'true',
    });
  }

  /** Send an RPC command and wait for a response. */
  @Get(':id')
  getOrder(@Param('id', ParseIntPipe) id: number): Observable<{ id: number; status: string }> {
    return this.client.send<{ id: number; status: string }>('order.get', { id });
  }
}

Use client.emit() for fire-and-forget events. Delivery is at-least-once through JetStream, and the caller gets no response back. Use client.send() for request/reply RPC, which returns an Observable<TResponse> with the handler's reply.

Broadcast prefix

To send a broadcast event, prefix the pattern with broadcast: when calling emit(). On the handler side, use { broadcast: true } in the decorator extras: no prefix needed.

Test it​

Start the application and trigger some messages:

# Start the app
npm run start:dev

# Send a workqueue event
curl http://localhost:3000/orders/create

# Send a broadcast event
curl http://localhost:3000/orders/broadcast

# Send an RPC command
curl http://localhost:3000/orders/42

What's next?​