0
# NestJS Microservices
1

2
NestJS Microservices is a comprehensive framework for building scalable distributed systems using various transport layers and communication patterns. It provides seamless integration with multiple messaging protocols including Redis, RabbitMQ, MQTT, NATS, gRPC, and Kafka, offering both request-response and event-driven communication models with automatic message serialization, pattern-based routing, and built-in error handling.
3

4
## Package Information
5

6
- **Package Name**: @nestjs/microservices
7
- **Package Type**: npm
8
- **Language**: TypeScript
9
- **Installation**: `npm install @nestjs/microservices`
10
- **Peer Dependencies**: `@nestjs/common`, `@nestjs/core`, `rxjs`, `reflect-metadata`
11

12
## Core Imports
13

14
```typescript
15
import { 
16
  NestFactory,
17
  ClientProxy,
18
  Transport,
19
  MessagePattern,
20
  EventPattern,
21
  Payload,
22
  Ctx
23
} from '@nestjs/microservices';
24
```
25

26
For CommonJS:
27

28
```javascript
29
const { 
30
  NestFactory,
31
  ClientProxy,
32
  Transport,
33
  MessagePattern,
34
  EventPattern 
35
} = require('@nestjs/microservices');
36
```
37

38
## Basic Usage
39

40
### Creating a Microservice
41

42
```typescript
43
import { NestFactory } from '@nestjs/core';
44
import { Transport, MicroserviceOptions } from '@nestjs/microservices';
45
import { AppModule } from './app.module';
46

47
async function bootstrap() {
48
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(
49
    AppModule,
50
    {
51
      transport: Transport.TCP,
52
      options: {
53
        host: '127.0.0.1',
54
        port: 3001,
55
      },
56
    },
57
  );
58
  
59
  await app.listen();
60
}
61
bootstrap();
62
```
63

64
### Message Handlers
65

66
```typescript
67
import { Controller } from '@nestjs/common';
68
import { MessagePattern, EventPattern, Payload, Ctx } from '@nestjs/microservices';
69

70
@Controller()
71
export class AppController {
72
  @MessagePattern({ cmd: 'sum' })
73
  accumulate(@Payload() data: number[]): number {
74
    return (data || []).reduce((a, b) => a + b, 0);
75
  }
76

77
  @EventPattern('user_created')
78
  handleUserCreated(@Payload() data: Record<string, unknown>) {
79
    console.log('User created:', data);
80
  }
81
}
82
```
83

84
### Client Usage
85

86
```typescript
87
import { Injectable } from '@nestjs/common';
88
import { ClientProxy, ClientProxyFactory, Transport } from '@nestjs/microservices';
89

90
@Injectable()
91
export class AppService {
92
  private client: ClientProxy;
93

94
  constructor() {
95
    this.client = ClientProxyFactory.create({
96
      transport: Transport.TCP,
97
      options: {
98
        host: '127.0.0.1',
99
        port: 3001,
100
      },
101
    });
102
  }
103

104
  async getSum(data: number[]): Promise<number> {
105
    return this.client.send({ cmd: 'sum' }, data).toPromise();
106
  }
107

108
  emitEvent(data: any): void {
109
    this.client.emit('user_created', data);
110
  }
111
}
112
```
113

114
## Architecture
115

116
NestJS Microservices is built around several key components:
117

118
- **Transport Layer**: Abstracted communication layer supporting multiple protocols (TCP, Redis, NATS, MQTT, gRPC, RabbitMQ, Kafka)
119
- **Client-Server Pattern**: Separate client and server implementations for each transport with connection management
120
- **Context System**: Transport-specific context objects providing access to underlying protocol details
121
- **Decorator System**: Method and parameter decorators for message handling and payload extraction
122
- **Observable-Based**: Built on RxJS observables for reactive programming patterns
123
- **Pattern Matching**: Flexible pattern-based message routing with metadata support
124

125
## Capabilities
126

127
### Microservice Application
128

129
Core microservice application class providing lifecycle management, middleware registration, and server initialization for distributed applications.
130

131
```typescript { .api }
132
class NestMicroservice extends NestApplicationContext implements INestMicroservice {
133
  readonly status: Observable<any>;
134
  
135
  createServer(config: MicroserviceOptions | AsyncMicroserviceOptions): void;
136
  registerModules(): Promise<any>;
137
  registerListeners(): void;
138
  useWebSocketAdapter(adapter: WebSocketAdapter): this;
139
  useGlobalFilters(...filters: ExceptionFilter[]): this;
140
  useGlobalPipes(...pipes: PipeTransform<any>[]): this;
141
  useGlobalInterceptors(...interceptors: NestInterceptor[]): this;
142
  useGlobalGuards(...guards: CanActivate[]): this;
143
  init(): Promise<this>;
144
  listen(): Promise<any>;
145
  close(): Promise<any>;
146
  setIsInitialized(isInitialized: boolean): void;
147
  setIsTerminated(isTerminated: boolean): void;
148
  setIsInitHookCalled(isInitHookCalled: boolean): void;
149
  on(event: string | number | symbol, callback: Function): any;
150
  unwrap<T>(): T;
151
}
152
```
153

154
### Server Infrastructure
155

156
Abstract server base class and factory for creating transport-specific server implementations with message handling and lifecycle management.
157

158
```typescript { .api }
159
abstract class Server<EventsMap = Record<string, Function>, Status = string> {
160
  readonly status: Observable<Status>;
161
  transportId?: Transport | symbol;
162
  
163
  abstract on<EventKey, EventCallback>(event: EventKey, callback: EventCallback): any;
164
  abstract unwrap<T>(): T;
165
  abstract listen(callback: (...optionalParams: unknown[]) => any): any;
166
  abstract close(): any;
167
  setTransportId(transportId: Transport | symbol): void;
168
  setOnProcessingStartHook(hook: Function): void;
169
  setOnProcessingEndHook(hook: Function): void;
170
  addHandler(pattern: any, callback: MessageHandler, isEventHandler?: boolean, extras?: Record<string, any>): void;
171
  getHandlers(): Map<string, MessageHandler>;
172
  getHandlerByPattern(pattern: string): MessageHandler | null;
173
  send(stream$: Observable<any>, respond: (data: WritePacket) => Promise<unknown> | void): Subscription;
174
  handleEvent(pattern: string, packet: ReadPacket, context: BaseRpcContext): Promise<any>;
175
}
176

177
class ServerFactory {
178
  static create(microserviceOptions: MicroserviceOptions): Server;
179
}
180
```
181

182
### Record Builders
183

184
Message record builders for constructing transport-specific message wrappers with metadata and options for enhanced message handling.
185

186
```typescript { .api }
187
class RmqRecord<TData = any> {
188
  constructor(data: TData, options?: RmqRecordOptions);
189
  readonly data: TData;
190
  options?: RmqRecordOptions;
191
}
192

193
class RmqRecordBuilder<TData> {
194
  constructor(data?: TData);
195
  setOptions(options: RmqRecordOptions): this;
196
  setData(data: TData): this;
197
  build(): RmqRecord;
198
}
199

200
class MqttRecord<TData = any> {
201
  constructor(data: TData, options?: MqttRecordOptions);
202
  readonly data: TData;
203
  options?: MqttRecordOptions;
204
}
205

206
class MqttRecordBuilder<TData> {
207
  constructor(data?: TData);
208
  setData(data: TData): this;
209
  setQoS(qos: 0 | 1 | 2): this;
210
  setRetain(retain: boolean): this;
211
  setDup(dup: boolean): this;
212
  setProperties(properties: Record<string, any>): this;
213
  build(): MqttRecord;
214
}
215

216
class NatsRecord<TData = any, THeaders = any> {
217
  constructor(data: TData, headers?: THeaders);
218
  readonly data: TData;
219
  readonly headers?: THeaders;
220
}
221

222
class NatsRecordBuilder<TData> {
223
  constructor(data?: TData);
224
  setHeaders<THeaders = any>(headers: THeaders): this;
225
  setData(data: TData): this;
226
  build(): NatsRecord;
227
}
228
```
229

230
### Client Proxies
231

232
Client-side communication interface for sending messages and events to microservices with support for multiple transport protocols.
233

234
```typescript { .api }
235
abstract class ClientProxy<EventsMap extends Record<never, Function> = Record<never, Function>, Status extends string = string> {
236
  connect(): Promise<any>;
237
  close(): any;
238
  send<TResult = any, TInput = any>(
239
    pattern: any,
240
    data: TInput
241
  ): Observable<TResult>;
242
  emit<TResult = any, TInput = any>(
243
    pattern: any,
244
    data: TInput
245
  ): Observable<TResult>;
246
}
247

248
class ClientProxyFactory {
249
  static create(clientOptions: ClientOptions): ClientProxy;
250
}
251
```
252

253
[Client Proxies](./client-proxies.md)
254

255
### Transport Implementations
256

257
Support for multiple messaging protocols including TCP, Redis, NATS, MQTT, gRPC, RabbitMQ, and Kafka with protocol-specific features and configurations.
258

259
```typescript { .api }
260
enum Transport {
261
  TCP = 0,
262
  REDIS = 1,
263
  NATS = 2,
264
  MQTT = 3,
265
  GRPC = 4,
266
  RMQ = 5,
267
  KAFKA = 6
268
}
269

270
interface MicroserviceOptions {
271
  transport?: Transport | CustomTransportStrategy;
272
  options?: any;
273
}
274
```
275

276
[Transport Implementations](./transports.md)
277

278
### Message Patterns & Event Handling
279

280
Decorators and patterns for defining message handlers with request-response and event-driven communication models.
281

282
```typescript { .api }
283
function MessagePattern<T = string>(
284
  metadata?: T,
285
  transport?: Transport | symbol,
286
  extras?: Record<string, any>
287
): MethodDecorator;
288

289
function EventPattern<T = string>(
290
  metadata?: T,
291
  transport?: Transport | symbol,
292
  extras?: Record<string, any>
293
): MethodDecorator;
294

295
function Payload(property?: string, ...pipes: PipeTransform[]): ParameterDecorator;
296
function Ctx(): ParameterDecorator;
297
```
298

299
[Message Patterns & Events](./message-patterns.md)
300

301
### Context Objects
302

303
Transport-specific context objects providing access to underlying protocol details and metadata for message handling.
304

305
```typescript { .api }
306
abstract class BaseRpcContext<T = any[]> {
307
  getArgs(): T;
308
  getArgByIndex<T = any>(index: number): T;
309
}
310

311
class KafkaContext extends BaseRpcContext {
312
  getMessage(): any;
313
  getPartition(): number;
314
  getTopic(): string;
315
  getConsumer(): any;
316
  getProducer(): any;
317
}
318
```
319

320
[Context Objects](./context-objects.md)
321

322
### gRPC Integration
323

324
Specialized support for gRPC protocol with service definitions, streaming methods, and protocol buffer integration.
325

326
```typescript { .api }
327
function GrpcMethod(service?: string, method?: string): MethodDecorator;
328
function GrpcStreamMethod(service?: string, method?: string): MethodDecorator;
329
function GrpcStreamCall(service?: string, method?: string): MethodDecorator;
330

331
interface ClientGrpc {
332
  getService<T = any>(name: string): T;
333
  getClientByServiceName<T = any>(name: string): T;
334
}
335
```
336

337
[gRPC Integration](./grpc.md)
338

339
### Exception Handling
340

341
Microservice-specific exception classes and error handling patterns for distributed system error management.
342

343
```typescript { .api }
344
class RpcException extends Error {
345
  constructor(error: string | object);
346
  getError(): string | object;
347
}
348

349
class KafkaRetriableException extends RpcException {
350
  constructor(error: string | object);
351
}
352
```
353

354
[Exception Handling](./exceptions.md)
355

356
### Module Integration
357

358
NestJS module integration for dependency injection and configuration management of microservice clients and servers.
359

360
```typescript { .api }
361
class ClientsModule {
362
  static register(options: ClientsModuleOptions): DynamicModule;
363
  static registerAsync(options: ClientsModuleAsyncOptions): DynamicModule;
364
}
365

366
function Client(metadata?: ClientOptions): PropertyDecorator;
367
```
368

369
[Module Integration](./modules.md)
370

371
## Types
372

373
```typescript { .api }
374
interface ClientOptions {
375
  transport: Transport | CustomTransportStrategy;
376
  options?: any;
377
}
378

379
interface CustomTransportStrategy {
380
  server: any;
381
  client: any;
382
}
383

384
interface ReadPacket<T = any> {
385
  pattern: any;
386
  data: T;
387
}
388

389
interface WritePacket<T = any> {
390
  err?: any;
391
  response?: T;
392
  isDisposed?: boolean;
393
  status?: string;
394
}
395

396
interface PacketId {
397
  id: string;
398
}
399

400
type OutgoingRequest = ReadPacket & PacketId;
401
type IncomingRequest = ReadPacket & PacketId;
402
type OutgoingEvent = ReadPacket;
403
type IncomingEvent = ReadPacket;
404
type IncomingResponse = WritePacket & PacketId;
405
type OutgoingResponse = WritePacket & PacketId;
406

407
interface RmqRecordOptions {
408
  expiration?: string | number;
409
  userId?: string;
410
  CC?: string | string[];
411
  mandatory?: boolean;
412
  persistent?: boolean;
413
  deliveryMode?: boolean | number;
414
  BCC?: string | string[];
415
  contentType?: string;
416
  contentEncoding?: string;
417
  headers?: Record<string, string>;
418
  priority?: number;
419
  messageId?: string;
420
  timestamp?: number;
421
  type?: string;
422
  appId?: string;
423
}
424

425
interface MqttRecordOptions {
426
  qos?: 0 | 1 | 2;
427
  retain?: boolean;
428
  dup?: boolean;
429
  properties?: {
430
    payloadFormatIndicator?: boolean;
431
    messageExpiryInterval?: number;
432
    topicAlias?: number;
433
    responseTopic?: string;
434
    correlationData?: Buffer;
435
    userProperties?: Record<string, string | string[]>;
436
    subscriptionIdentifier?: number;
437
    contentType?: string;
438
  };
439
}
440
```