API

import { Module } from '@nestjs/common';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { KubeMQModule } from '@kubemq/nestjs-transport';

@Module({
  imports: [
    // Shared connection (global by default)
    KubeMQModule.forRoot({
      address: 'localhost:50000',
      clientId: 'orders-app',
      isGlobal: true,
    }),

    // Async shared connection from ConfigService
    KubeMQModule.forRootAsync({
      imports: [ConfigModule],
      useFactory: (config: ConfigService) => ({
        address: config.getOrThrow('KUBEMQ_ADDRESS'),
        clientId: config.get('KUBEMQ_CLIENT_ID'),
      }),
      inject: [ConfigService],
    }),

    // Named client proxy for injection
    KubeMQModule.register({
      name: 'ORDER_SERVICE',
      address: 'localhost:50000',
      clientId: 'order-client',
    }),

    // Channel-scoped client: every channel is prefixed with "orders"
    KubeMQModule.forFeature({ name: 'ORDER_BUS', channelPrefix: 'orders' }),
  ],
})
export class AppModule {}

import {
  CommandHandler,
  QueryHandler,
  EventHandler,
  EventStoreHandler,
  QueueHandler,
} from '@kubemq/nestjs-transport';
import { Payload, Ctx } from '@nestjs/microservices';

@CommandHandler('orders.create', { group: 'order-writers', maxConcurrent: 5 })
handleCreate(@Payload() data: CreateOrderDto) { /* ... */ }

@QueryHandler('orders.get')
handleGet(@Payload() data: { id: string }) { /* ... */ }

// A single handler can subscribe to multiple channels
@EventHandler(['orders.updated', 'orders.created'])
onChange(@Payload() data: OrderEvent) { /* ... */ }

@EventStoreHandler('orders.history', { startFrom: 'first' })
replay(@Payload() data: unknown) { /* ... */ }

@QueueHandler('orders.process', { manualAck: true, maxMessages: 5 })
process(@Payload() data: unknown, @Ctx() ctx) { /* ... */ }

// Replay an events-store stream from sequence 100
@EventStoreHandler('orders.history', { startFrom: 'sequence', startValue: 100 })

// Manual-ack queue with bounded concurrency and a DLQ after 3 retries
@QueueHandler('orders.process', {
  manualAck: true,
  maxMessages: 5,
  waitTimeoutSeconds: 60,
  concurrency: 4,
  deadLetterChannel: 'orders.dlq',
  maxRetries: 3,
})

// Batch processing with idempotent dedup
@QueueHandler('orders.batch', {
  batch: true,
  maxMessages: 10,
  idempotency: { ttlSeconds: 600, maxCacheSize: 50_000 },
})

import {
  EventStoreHandler,
  QueueHandler,
  KubeMQEventStoreContext,
  KubeMQQueueContext,
} from '@kubemq/nestjs-transport';
import { Payload, Ctx } from '@nestjs/microservices';

@EventStoreHandler('orders.history', { startFrom: 'first' })
onHistory(@Payload() data: unknown, @Ctx() ctx: KubeMQEventStoreContext) {
  console.log(`Event #${ctx.sequence} on ${ctx.channel}`);
}

@QueueHandler('orders.process', { manualAck: true })
async onProcess(@Payload() data: unknown, @Ctx() ctx: KubeMQQueueContext) {
  console.log(`Delivery #${ctx.receiveCount}, re-routed: ${ctx.isReRouted}`);
  try {
    await handle(data);
    ctx.ack();
  } catch {
    ctx.reQueue('orders.dlq'); // or ctx.nack()
  }
}

import { KubeMQServer, MessagePackSerializer, MessagePackDeserializer } from '@kubemq/nestjs-transport';

new KubeMQServer({
  address: 'localhost:50000',
  serializer: new MessagePackSerializer(),
  deserializer: new MessagePackDeserializer(),
});