\n\n# Table of Contents\n\n- [@golevelup/nestjs-rabbitmq](#golevelupnestjs-rabbitmq)\n- [Table of Contents](#table-of-contents)\n - [Description](#description)\n - [Motivation](#motivation)\n - [Connection Management](#connection-management)\n - [Usage](#usage)\n - [Install](#install)\n - [Module Initialization](#module-initialization)\n - [Usage with Interceptors](#usage-with-interceptors)\n - [Usage with Controllers](#usage-with-controllers)\n - [Receiving Messages](#receiving-messages)\n - [Exposing RPC Handlers](#exposing-rpc-handlers)\n - [Exposing Pub/Sub Handlers](#exposing-pubsub-handlers)\n - [Message Handling](#message-handling)\n - [Conditional Handler Registration](#conditional-handler-registration)\n - [Selecting channel for handler](#selecting-channel-for-handler)\n - [Sending Messages](#sending-messages)\n - [Inject the AmqpConnection](#inject-the-amqpconnection)\n - [Publising Messages (Fire and Forget)](#publising-messages-fire-and-forget)\n - [Requesting Data from an RPC](#requesting-data-from-an-rpc)\n - [Type Inference](#type-inference)\n - [Interop with other RPC Servers](#interop-with-other-rpc-servers)\n - [Advanced Patterns](#advanced-patterns)\n - [Competing Consumers](#competing-consumers)\n - [Contribute](#contribute)\n - [License](#license)\n\n## Description\n\nThis module features an opinionated set of decorators for common RabbitMQ patterns including Publish/Subscribe and RPC using Rabbit's [Direct Reply-To Queue](https://www.rabbitmq.com/direct-reply-to.html) for optimal performance.\n\nIt allows you to expose normal NestJS service methods as messaging handlers that can be configured to support a variety of messaging patterns.\n\n## Motivation\n\nNestJS offers an out of the box microservices experience with support for a variety of transports. However, because NestJS microservices strives to work with a variety of transport mechanisms in a generic way, it misses out on some of the powerful functionality offered by individual transport layers.\n\nSome of the most notable missing functionality includes common messaging patterns like publish/subscribe and competing consumers.\n\n## Connection Management\n\nIn previous versions, this package did not support advanced connection management and if you tried to launch the app when a connection could not be established, an error was thrown and caused the app to crash.\n\nNow, this package leverages [`amqp-connection-manager`](https://github.com/benbria/node-amqp-connection-manager) package to support connection resiliency.\n\n**NOTE**: to maintain the same previous behavior and not introduce a major version update, the previous behavior is still the default.\n\nIf you want to transition to the new behavior and enable connection resiliency, you can configure `connectionInitOptions` to not wait for a connection to be availble, for example:\n\n```typescript\nimport { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';\n\n@Module({\n imports: [\n RabbitMQModule.forRoot(RabbitMQModule, {\n exchanges: [\n {\n name: 'exchange1',\n type: 'topic',\n },\n ],\n uri: 'amqp://rabbitmq:rabbitmq@localhost:5672',\n connectionInitOptions: { wait: false },\n }),\n ],\n})\nexport class RabbitExampleModule {}\n```\n\nWith the new behavior in place, unavailability of a RabbitMQ broker still allows your application to bootstrap correctly and relevant channel setups take place whenever a connection can be established.\n\nThe same principle applies to when a connection is lost. In such cases, the module tries to reconnect and set up everything again once it is reconnected.\n\n## Usage\n\n### Install\n\n`npm install ---save @golevelup/nestjs-rabbitmq`\n\nor\n\n`yarn add @golevelup/nestjs-rabbitmq`\n\n### Module Initialization\n\nImport and add `RabbitMQModule` it to the `imports` array of module for which you would like to discover handlers. It may make sense for your application to do this in a shared module or to re-export it so it can be used across modules more easily. [Refer to the NestJS docs on modules for more information.](https://docs.nestjs.com/modules)\n\nIf you are using exchanges, provide information about them to the module and they will be automatically asserted for you as part of initialization. If you don't, it's possible message passing will fail if an exchange is addressed that hasn't been created yet.\n\nYou can also optionally create your own channels which you consume messages from. If you don't create your own channels there will always be one created by default. You can also select which channel is default if you are creating your own. By setting `prefetchCount` for a particular channel you can manage message speeds of your various handlers on the same connection.\n\n```typescript\nimport { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';\nimport { Module } from '@nestjs/common';\nimport { MessagingController } from './messaging/messaging.controller';\nimport { MessagingService } from './messaging/messaging.service';\n\n@Module({\n imports: [\n RabbitMQModule.forRoot(RabbitMQModule, {\n exchanges: [\n {\n name: 'exchange1',\n type: 'topic',\n },\n ],\n uri: 'amqp://rabbitmq:rabbitmq@localhost:5672',\n channels: {\n 'channel-1': {\n prefetchCount: 15,\n default: true,\n },\n 'channel-2': {\n prefetchCount: 2,\n },\n },\n }),\n RabbitExampleModule,\n ],\n providers: [MessagingService],\n controllers: [MessagingController],\n})\nexport class RabbitExampleModule {}\n```\n\n## Usage with Interceptors\n\nThis library is built using an underlying NestJS concept called `External Contexts` which allows for methods to be included in the NestJS lifecycle. This means that Guards and Interceptors can be used in conjunction with RabbitMQ message handlers. However, this can have unwanted/unintended consequences if you are using Global intereceptors in your application as these will also apply to all RabbitMQ message handlers. As a workaround, there is a utiltity function available called `isRabbitContext` which you can use inside of Interceptors to do conditional logic.\n\n```typescript\nimport { isRabbitContext } from '@golevelup/nestjs-rabbitmq';\n\n@Injectable()\nclass ExampleInterceptor implements NestInterceptor {\n intercept(context: ExecutionContext, next: CallHandler) {\n const shouldSkip = isRabbitContext(context);\n if (shouldSkip) {\n return next.handle();\n }\n\n // Execute custom interceptor logic for HTTP request/response\n return next.handle();\n }\n}\n```\n\n## Usage with Controllers\n\nTo improve the migration process, it is possible to use NestJS controllers as handlers.\nWARNING: When using controllers, be aware that no HTTP context is available.\n\nTo enable the controller discovery the option enableControllerDiscovery has to be true.\n\n```typescript\nimport { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';\nimport { Module } from '@nestjs/common';\nimport { MessagingController } from './messaging/messaging.controller';\nimport { MessagingService } from './messaging/messaging.service';\n\n@Module({\n imports: [\n RabbitMQModule.forRoot(RabbitMQModule, {\n exchanges: [\n {\n name: 'exchange1',\n type: 'topic',\n },\n ],\n uri: 'amqp://rabbitmq:rabbitmq@localhost:5672',\n enableControllerDiscovery: true,\n }),\n RabbitExampleModule,\n ],\n providers: [MessagingService, MessagingController],\n controllers: [MessagingController],\n})\nexport class RabbitExampleModule {}\n```\n\n### Interceptors, Guards, Pipes\n\nTo use Interceptors, Guards or Pipes, the controller has to be imported as provider in the module.\nThen simly add the corresponding decorator to the whole controller or the method.\n\n```typescript\n@RabbitRPC({\n routingKey: 'intercepted-rpc-2',\n exchange: 'exchange2',\n queue: 'intercepted-rpc-2',\n})\n@UseInterceptors(TransformInterceptor)\ninterceptedRpc() {\n return {\n message: 42,\n };\n}\n```\n\n## Receiving Messages\n\n### Exposing RPC Handlers\n\nSimply apply the `RabbitRPC` decorator to a new or existing NestJS service class. When a message matching the exchange and routing key is received over RabbitMQ, the result of the Service method will be automatically sent back to the requester using the [Direct Reply-To Queue](https://www.rabbitmq.com/direct-reply-to.html).\n\n```typescript\nimport { RabbitRPC } from '@golevelup/nestjs-rabbitmq';\nimport { Injectable } from '@nestjs/common';\n\n@Injectable()\nexport class MessagingService {\n @RabbitRPC({\n exchange: 'exchange1',\n routingKey: 'rpc-route',\n queue: 'rpc-queue',\n })\n public async rpcHandler(msg: {}) {\n return {\n response: 42,\n };\n }\n}\n```\n\n### Exposing Pub/Sub Handlers\n\nSimply apply the `RabbitSubscribe` decorator to a new or existing NestJS service class. When a message matching the exchange and routing key is received over RabbitMQ, the service method will automatically be invoked with the message allowing it to be handled as necessary.\n\n```typescript\nimport { RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';\nimport { Injectable } from '@nestjs/common';\n\n@Injectable()\nexport class MessagingService {\n @RabbitSubscribe({\n exchange: 'exchange1',\n routingKey: 'subscribe-route',\n queue: 'subscribe-queue',\n })\n public async pubSubHandler(msg: {}) {\n console.log(`Received message: ${JSON.stringify(msg)}`);\n }\n}\n```\n\n### Message Handling\n\nNestJS Plus provides sane defaults for message handling with automatic acking of messages that have been successfully processed by either RPC or PubSub handlers. However, there are situtations where an application may want to Negatively Acknowledge (or Nack) a message. To support this, the library exposes the `Nack` object which when returned from a handler allows a developer to control the message handling behavior. Simply return a `Nack` instance to negatively acknowledge the message.\n\nBy default, messages that are Nacked will not be requeued. However, if you would like to requeue the message so that another handler has an opportunity to process it use the optional requeue constructor argument set to true.\n\n```typescript\nimport { RabbitRPC } from '@golevelup/nestjs-rabbitmq';\nimport { Injectable } from '@nestjs/common';\n\n@Injectable()\nexport class MessagingService {\n @RabbitRPC({\n exchange: 'exchange1',\n routingKey: 'rpc-route',\n queue: 'rpc-queue'\n })\n public async rpcHandler(msg: {}) {\n return {\n if (someCondition) {\n return 42;\n } else if (requeueCondition) {\n return new Nack(true);\n } else {\n // Will not be requeued\n return new Nack();\n }\n };\n }\n}\n```\n\n### Conditional Handler Registration\n\nIn some scenarios, it may not be desirable for all running instances of a NestJS application to register RabbitMQ message handlers. For example, if leveraging the same application code base to expose API instances and worker roles separately it may be desirable to have only the worker instances attach handlers to manage queue subscriptions or RPC requests.\n\nThe default behavior is that handlers will be attached, but to opt out simply set the `registerHandlers` configuration option to `false` when registering the RabbitMQModule.\n\n### Dealing with the amqp original message\n\nIn some scenarios, it wil be useful to get the original amqp message (to retrieve the fields, properties...).\n\nThe raw message is passed to the consumer as a second argument.\n\nIf the method signature of the consumer accepts `amqplib.ConsumeMessage` as a second argument, it enables to access all information that is available on the original message.\n\n```typescript\nimport { RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';\nimport { Injectable } from '@nestjs/common';\nimport { ConsumeMessage } from 'amqplib';\n\n@Injectable()\nexport class MessagingService {\n @RabbitSubscribe({\n exchange: 'exchange1',\n routingKey: 'subscribe-route',\n queue: 'subscribe-queue',\n })\n public async pubSubHandler(msg: {}, amqpMsg: ConsumeMessage) {\n console.log(`Correlation id: ${amqpMsg.properties.correlationId}`);\n }\n}\n```\n\n### Selecting channel for handler\n\nYou can optionally select channel which handler uses to consume messages from.\n\nSet the `queueOptions.channel` to the name of the channel to enable this feature. If channel does not exist or you haven't specified one, it will use the default channel. For channel to exist it needs to be created in module config.\n\n```typescript\nimport { RabbitSubscribe, RabbitRPC } from '@golevelup/nestjs-rabbitmq';\nimport { Injectable } from '@nestjs/common';\n\n@Injectable()\nexport class MessagingService {\n @RabbitRPC({\n exchange: 'exchange1',\n routingKey: 'subscribe-route',\n queue: 'subscribe-queue',\n queueOptions: {\n channel: 'channel-2',\n },\n })\n public async rpcHandler(msg: {}) {\n console.log(`Received rpc message: ${JSON.stringify(msg)}`);\n\n return { message: 'hi' };\n }\n\n @RabbitSubscribe({\n exchange: 'exchange1',\n routingKey: 'subscribe-route-2',\n queue: 'subscribe-queue-2',\n })\n public async pubSubHandler(msg: {}) {\n console.log(`Received pub/sub message: ${JSON.stringify(msg)}`);\n }\n}\n```\n\n## Sending Messages\n\n### Inject the AmqpConnection\n\nAll RabbitMQ interactions go through the `AmqpConnection` object. Assuming you installed and configured the `RabbitMQModule`, the object can be obtained through Nest's dependency injection system. Simply require it as a constructor parameter in a Nest Controller or Service.\n\n```typescript\n@Controller()\nexport class AppController {\n constructor(private readonly amqpConnection: AmqpConnection) {}\n\n ...\n}\n```\n\n### Publising Messages (Fire and Forget)\n\nIf you just want to publish a message onto a RabbitMQ exchange, use the `publish` method of the `AmqpConnection` which has the following signature:\n\n```typescript\npublic publish(\n exchange: string,\n routingKey: string,\n message: any,\n options?: amqplib.Options.Publish\n)\n```\n\nFor example:\n\n```typescript\namqpConnection.publish('some-exchange', 'routing-key', { msg: 'hello world' });\n```\n\n### Requesting Data from an RPC\n\nIf you'd like to request data from another RPC handler that's been set up using this library, you can use the `request` method of the `AmqpConnection`.\n\nFor example:\n\n```typescript\nconst response = await amqpConnection.request({\n exchange: 'exchange1',\n routingKey: 'rpc',\n payload: {\n request: 'val',\n },\n timeout: 10000, // optional timeout for how long the request\n // should wait before failing if no response is received\n});\n```\n\n#### Type Inference\n\nThe generic parameter used with the `request` method lets you specify the _expected_ return type of the RPC response. This is useful for getting intellisense in your editor but no object validation of the actual received object is done on your behalf. This means that you are required to provide your own object validation logic if you need to make runtime guarantees about message structure\n\n#### Interop with other RPC Servers\n\nThe RPC functionality included in `@golevelup/nestjs-rabbitmq` is based on the [Direct Reply-To Queue](https://www.rabbitmq.com/direct-reply-to.html) functionality of RabbitMQ. It is possible that because of this, the client library (`AmqpConnection.request`) could be used to interact with an RPC server implemented using a different language or framework. However, this functionality has not been verified.\n\n## Advanced Patterns\n\n### Competing Consumers\n\nThe competing consumer pattern is useful when building decoupled applications especially when it comes to things like RPC or [Work Queues](https://www.rabbitmq.com/tutorials/tutorial-two-javascript.html). In these scenarios, it often desirable to ensure that only one handler processes a given message especially if your app is horizontally scaled.\n\nIn the previous examples, both RPC and Pub/Sub would be using the Competing Consumer pattern by default through the use of a named `queue` parameter. If running multiple instances of the application, each instance would bind to the same named queue and receive the messages in a round robin fashion.\n\nIf you don't want this behavior, simply don't provide a queue name. A unique one will be generated automatically and all instances of the handler will receive their own copy of the message.\n\n**Important** RPC behavior has not been tested without the use of a named queue as this would cause multiple messages to potentially be sent back in response to a single request. If you're using RPC it is highly recommended that you specify a named queue. The API may be updated in the future to specifically require this.\n\n```typescript\nimport { RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';\nimport { Injectable } from '@nestjs/common';\n\n@Injectable()\nexport class MessagingService {\n @RabbitSubscribe({\n exchange: 'exchange1',\n routingKey: 'subscribe-route1',\n queue: 'subscribe-queue',\n })\n public async competingPubSubHandler(msg: {}) {\n console.log(`Received message: ${JSON.stringify(msg)}`);\n }\n\n @RabbitSubscribe({\n exchange: 'exchange1',\n routingKey: 'subscribe-route2',\n })\n public async messagePerInstanceHandler(msg: {}) {\n console.log(`Received message: ${JSON.stringify(msg)}`);\n }\n}\n```\n\n### Handling errors\n\nBy default, the library tries to do its best to give you the control on errors if you want and to do something sensible by default.\n\nThis is done with the `errorHandler` property that is availble both in RPC and RabbitSubscribe.\n\n```typescript\n @RabbitSubscribe({\n exchange: 'exchange1',\n routingKey: 'subscribe-route1',\n queue: 'subscribe-queue',\n errorHandler: myErrorHandler\n })\n```\n\n> it should be used with `rpcOptions` for RPC\n\nThe default is `defaultNackErrorHandler` and it just nack the message without requeue (which is usually ok to avoid the message coming back in the queue again and again)\n\nHowever, you can do more fancy stuff like inspecting the message properties to decide to requeue or not. Be aware that you should not requeue indefinitely...\n\nPlease note that nack will trigger the dead-letter mecanism of RabbitMQ (and so, you can use the deadLetterExchange in the queueOptions in order to send the message somewhere else).\n\nA complete error handling strategy for RabbitMQ is out of the scope of this library.\n\n### Handling errors during queue creation\n\nSimilarly to message errors, the library provide an error handler for failures during a queue creation (more exactly, during the assertQueue operation which will create the queue if it does not exist).\n\n```typescript\n @RabbitSubscribe({\n exchange: 'exchange1',\n routingKey: 'subscribe-route1',\n queue: 'subscribe-queue',\n assertQueueErrorHandler: myErrorHandler\n })\n```\n\nThe default is `defaultAssertQueueErrorHandler` which just rethrows the RabbitMq error (because there is no \"one size fits all\" for this situation).\n\nYou have the option to use `forceDeleteAssertQueueErrorHandler` which will try to delete the queue and recreate it with the provided queueOptions (if any)\n\nObviously, you can also provide your own function and do whatever is best for you, in this case the function must return the name of the created queue.\n\n## Contribute\n\nContributions welcome! Read the [contribution guidelines](../../CONTRIBUTING.md) first.\n\n## License\n\n[MIT License](../../LICENSE)\n","readmeFilename":"README.md"}
![\"version\"](\"https://img.shields.io/npm/v/@golevelup/nestjs-rabbitmq.svg?style=flat\")
\n![\"downloads\"](\"https://img.shields.io/npm/dt/@golevelup/nestjs-rabbitmq.svg?style=flat\")
\n![\"license\"](\"https://img.shields.io/npm/l/@golevelup/nestjs-rabbitmq.svg\")
\n