Golevelup NestJS RabbitMQ

https://golevelup.github.io/nestjs/modules/rabbitmq.html
Admin
A NestJS module providing opinionated decorators for RabbitMQ messaging patterns including Pub/Sub,...
Tokens:9,711
Snippets:75
Trust Score:8.3
Update:1 week ago
Show doc for...
Tokens:
Raw
### Install with npm

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Install the @golevelup/nestjs-rabbitmq package using npm.

```bash
npm install ---save @golevelup/nestjs-rabbitmq
```

--------------------------------

### Install with Yarn

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Install the @golevelup/nestjs-rabbitmq package using Yarn.

```bash
yarn add @golevelup/nestjs-rabbitmq
```

--------------------------------

### Install with pnpm

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Install the @golevelup/nestjs-rabbitmq package using pnpm.

```bash
pnpm add @golevelup/nestjs-rabbitmq
```

--------------------------------

### Enable connection resiliency with connectionInitOptions

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Configure RabbitMQModule with connectionInitOptions set to { wait: false } to enable connection resiliency. This allows the application to bootstrap even if the RabbitMQ broker is unavailable, and channel setups occur when a connection is established.

```typescript
import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';

@Module({
  imports: [
    RabbitMQModule.forRoot({
      exchanges: [
        {
          name: 'exchange1',
          type: 'topic',
        },
      ],
      uri: 'amqp://rabbitmq:rabbitmq@localhost:5672',
      connectionInitOptions: { wait: false },
    }),
  ],
})
export class RabbitExampleModule {}
```

--------------------------------

### Read retry counter from message headers for conditional Nack

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Demonstrates reading a retry counter from message headers to conditionally decide between requeueing and dead-lettering. The x-retry-count header must be set by your retry infrastructure when republishing; this example only shows the consumer side.

```typescript
import { RabbitSubscribe, Nack } from '@golevelup/nestjs-rabbitmq';
import { Injectable } from '@nestjs/common';
import { ConsumeMessage } from 'amqplib';

@Injectable()
export class MessagingService {
  @RabbitSubscribe({
    exchange: 'exchange1',
    routingKey: 'subscribe-route',
    queue: 'subscribe-queue',
  })
  public async pubSubHandler(msg: {}, amqpMsg: ConsumeMessage) {
    // x-retry-count must be set by your retry infrastructure when republishing
    const retryCount = (amqpMsg.properties.headers?.['x-retry-count'] ?? 0) as number;

```

--------------------------------

### Custom Message Parsers for Non-JSON Formats

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Override default JSON.parse/JSON.stringify behavior by providing custom parsers to RabbitMQModule. The example shows imports and module setup.

```typescript
import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
import { Module } from '@nestjs/common';
import { MessagingController } from './messaging/messaging.controller';
import { MessagingService } from './messaging/messaging.service';
import { ConsumeMessage } from 'amqplib';
```

--------------------------------

### Initialize RabbitMQModule with exchanges and channels

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Initializes RabbitMQModule with exchange definitions, connection URI, and custom channels. Exchanges are automatically asserted during initialization; if omitted, message passing may fail when addressing an uncreated exchange. Custom channels with prefetchCount manage message speeds; if none are created, a default channel is used.

```typescript
import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
import { Module } from '@nestjs/common';
import { MessagingController } from './messaging/messaging.controller';
import { MessagingService } from './messaging/messaging.service';

@Module({
  imports: [
    RabbitMQModule.forRoot({
      exchanges: [
        {
          name: 'exchange1',
          type: 'topic',
        },
      ],
      uri: 'amqp://rabbitmq:rabbitmq@localhost:5672',
      channels: {
        'channel-1': {
          prefetchCount: 15,
          default: true,
        },
        'channel-2': {
          prefetchCount: 2,
        },
      },
    }),
    RabbitExampleModule,
  ],
  providers: [MessagingService],
  controllers: [MessagingController],
})
export class RabbitExampleModule {}
```

--------------------------------

### Selecting channel for handler with queueOptions.channel

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Shows how to set the `queueOptions.channel` property on `@RabbitRPC` and `@RabbitSubscribe` decorators to select a specific channel for consuming messages. If the channel does not exist or is not specified, the default channel is used; the channel must be created in the module config.

```typescript
import { RabbitSubscribe, RabbitRPC } from '@golevelup/nestjs-rabbitmq';
import { Injectable } from '@nestjs/common';

@Injectable()
export class MessagingService {
  @RabbitRPC({
    exchange: 'exchange1',
    routingKey: 'subscribe-route',
    queue: 'subscribe-queue',
    queueOptions: {
      channel: 'channel-2',
    },
  })
  public async rpcHandler(msg: {}) {
    console.log(`Received rpc message: ${JSON.stringify(msg)}`);

    return { message: 'hi' };
  }

  @RabbitSubscribe({
    exchange: 'exchange1',
    routingKey: 'subscribe-route-2',
    queue: 'subscribe-queue-2',
  })
  public async pubSubHandler(msg: {}) {
    console.log(`Received pub/sub message: ${JSON.stringify(msg)}`);
  }
}
```

--------------------------------

### Configuring consumer-side message batching with batchOptions

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Configures consumer-side message batching with `batchOptions` on a `@RabbitSubscribe` decorator. Messages accumulate until the batch `size` limit or `timeout` expires, then all messages are acked or nacked automatically. The channel's `prefetchCount` must be >= batch `size`; an optional `errorHandler` can be provided, otherwise it falls back to the top-level `errorHandler` or default behavior.

```typescript
import { RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';

const batchErrorHandler = (channel, messages, error) => {
  console.log(`Received message batch of length: ${messages.length}`);
};

@Injectable()
export class MessagingService {
  @RabbitSubscribe({
    exchange: 'exchange1',
    routingKey: 'batch-route',
    queue: 'batch-queue',
    batchOptions: {
      size: 10,
      timeout: 200,
      errorHandler: batchErrorHandler,
    },
  })
  public async batchHandler(messages) {
    console.log(`Received message batch of length: ${messages.length}`);
  }
}
```

--------------------------------

### Mock AmqpConnection for testing with overrideModule

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Mocks the `AmqpConnection` provider in tests by creating a `MockRabbitExampleModule` and overriding the real `RabbitExampleModule` with it. This allows publishers that inject `AmqpConnection` to work without a real connection.

```typescript
// In your test setup...

import { Module } from '@nestjs/common';
import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
import { createMock } from '@golevelup/ts-jest';
import { Test } from '@nestjs/testing';
import { AppModule } from 'where your root module is located';
import { RabbitExampleModule } from 'where your rabbitmq module is located';

// Create a valid mock module
@Module({
  providers: [
    {
      provide: AmqpConnection,
      useValue: createMock<AmqpConnection>(),
    },
  ],
  exports: [AmqpConnection],
})
class MockRabbitExampleModule {}

// Then override the real `RabbitMqModule` with the mocked one
beforeAll(async () => {
  const moduleFixture = await Test.createTestingModule({
    imports: [AppModule],
  })
    .overrideModule(RabbitExampleModule)
    .useModule(MockRabbitExampleModule)
    .compile();

  const app = moduleFixture.createNestApplication();

  await app.init();
});
```

--------------------------------

### Configuring per-handler consumer tag via queueOptions.consumerOptions

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Sets a per-handler `consumerTag` via `queueOptions.consumerOptions` on a `@RabbitSubscribe` decorator. A tag set on the decorator takes precedence over the global queue config. Consumer tags must be unique per channel; duplicate tags cause RabbitMQ to reject the registration.

```typescript
import { RabbitSubscribe, RabbitRPC } from '@golevelup/nestjs-rabbitmq';
import { Injectable } from '@nestjs/common';

@Injectable()
export class MessagingService {
  @RabbitSubscribe({
    exchange: 'exchange1',
    routingKey: 'subscribe-route',
    queue: 'my-queue',
    queueOptions: {
      consumerOptions: {
        consumerTag: 'my-handler-consumer-tag',
      },
    },
  })
  public async pubSubHandler(msg: {}) {
    console.log(`Received message: ${JSON.stringify(msg)}`);
  }
}
```

--------------------------------

### Mix bindings with Top-Level Exchange and Routing Key

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Combine bindings with a top-level exchange and routingKey in RabbitSubscribe to bind a queue to multiple exchanges.

```typescript
@RabbitSubscribe({
  exchange: 'exchange1',
  routingKey: 'route.c',
  queue: 'my-queue',
  bindings: [
    { exchange: 'exchange2', routingKey: 'route.d' },
  ],
})
public async mixedHandler(msg: {}) {
  console.log(`Received message: ${JSON.stringify(msg)}`);
}
```

--------------------------------

### Configure defaultPublishOptions for persistence

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Configure defaultPublishOptions in RabbitMQModule.forRoot to apply persistent flag to all published messages by default.

```typescript
RabbitMQModule.forRoot(RabbitMQModule, {
  exchanges: [{ name: 'some-exchange', type: 'topic' }],
  uri: 'amqp://rabbitmq:rabbitmq@localhost:5672',
  defaultPublishOptions: {
    persistent: true,
  },
});
```

--------------------------------

### publish

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Publishes a message to a RabbitMQ exchange. This is a fire-and-forget operation.

```APIDOC
## publish

### Description
Publishes a message to a RabbitMQ exchange. This is a fire-and-forget operation.

### Method
publish

### Signature
```typescript
public publish(
  exchange: string,
  routingKey: string,
  message: any,
  options?: amqplib.Options.Publish
)
```

### Parameters
- **exchange** (string) - Required - The name of the exchange to publish to.
- **routingKey** (string) - Required - The routing key for the message.
- **message** (any) - Required - The message payload.
- **options** (amqplib.Options.Publish) - Optional - Additional publish options, such as `{ persistent: true }` to mark the message as persistent.

### Example
```typescript
amqpConnection.publish('some-exchange', 'routing-key', { msg: 'hello world' });
```

### Notes
- For messages to be retained across a broker restart, they must be published to durable queues/exchanges, the broker must flush them to disk, and the messages must be marked persistent.
- `defaultPublishOptions` can be configured in the module to apply persistence to all messages by default. Per-call options are merged on top of defaults.
```

--------------------------------

### Bind Queue to Multiple Exchanges with bindings

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Use the bindings option in RabbitSubscribe to bind a queue to multiple exchanges. Each entry specifies an exchange and routingKey, applied in addition to top-level options.

```typescript
import { RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
import { Injectable } from '@nestjs/common';

@Injectable()
export class MessagingService {
  @RabbitSubscribe({
    queue: 'my-queue',
    bindings: [
      { exchange: 'exchange1', routingKey: 'route.a' },
      { exchange: 'exchange2', routingKey: 'route.b' },
    ],
  })
  public async multiExchangeHandler(msg: {}) {
    console.log(`Received message: ${JSON.stringify(msg)}`);
  }
}
```

--------------------------------

### RabbitSubscribe competing consumers with and without named queue

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Shows two RabbitSubscribe handlers: one with a named queue for competing consumers (round-robin) and one without a queue name so each instance receives its own copy. Use the named queue for RPC to avoid multiple responses.

```typescript
import { RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
import { Injectable } from '@nestjs/common';

@Injectable()
export class MessagingService {
  @RabbitSubscribe({
    exchange: 'exchange1',
    routingKey: 'subscribe-route1',
    queue: 'subscribe-queue',
  })
  public async competingPubSubHandler(msg: {}) {
    console.log(`Received message: ${JSON.stringify(msg)}`);
  }

  @RabbitSubscribe({
    exchange: 'exchange1',
    routingKey: 'subscribe-route2',
  })
  public async messagePerInstanceHandler(msg: {}) {
    console.log(`Received message: ${JSON.stringify(msg)}`);
  }
}
```

--------------------------------

### Enable controller discovery with enableControllerDiscovery

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Enables controller discovery by setting enableControllerDiscovery to true in RabbitMQModule.forRoot. This allows NestJS controllers to be used as RabbitMQ handlers, but no HTTP context is available in that mode.

```typescript
import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
import { Module } from '@nestjs/common';
import { MessagingController } from './messaging/messaging.controller';
import { MessagingService } from './messaging/messaging.service';

@Module({
  imports: [
    RabbitMQModule.forRoot({
      exchanges: [
        {
          name: 'exchange1',
          type: 'topic',
        },
      ],
      uri: 'amqp://rabbitmq:rabbitmq@localhost:5672',
      enableControllerDiscovery: true,
    }),
    RabbitExampleModule,
  ],
  providers: [MessagingService, MessagingController],
  controllers: [MessagingController],
})
export class RabbitExampleModule {}
```

--------------------------------

### RpcTimeoutError handling

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Shows how to catch and handle RpcTimeoutError when an RPC request times out.

```APIDOC
## RpcTimeoutError

### Description
Thrown when an RPC request times out. Includes detailed information about the timeout.

### Usage
```typescript
import { AmqpConnection, RpcTimeoutError } from '@golevelup/nestjs-rabbitmq';

try {
  const response = await amqpConnection.request<ExpectedReturnType>({
    exchange: 'exchange1',
    routingKey: 'rpc',
    payload: { request: 'val' },
    timeout: 5000,
  });
} catch (error) {
  if (error instanceof RpcTimeoutError) {
    console.error(`RPC timed out after ${error.timeout}ms`);
    console.error(
      `Exchange: ${error.exchange}, Routing Key: ${error.routingKey}`,
    );
    // Handle timeout specifically
  } else {
    // Handle other errors
    throw error;
  }
}
```

### Properties
- **timeout** (number) - The timeout duration in milliseconds.
- **exchange** (string) - The exchange used in the request.
- **routingKey** (string) - The routing key used in the request.
```

--------------------------------

### Configuring global consumer tag in RabbitMQModule.forRoot

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Sets a default `consumerTag` for a queue in the module-level `queues` configuration. All handlers subscribed to that queue will use this tag unless overridden at the handler level.

```typescript
RabbitMQModule.forRoot({
  uri: 'amqp://localhost:5672',
  queues: [
    {
      name: 'my-queue',
      consumerTag: 'my-global-consumer-tag',
    },
  ],
});
```

--------------------------------

### Expose RPC Handler with RabbitRPC

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Apply RabbitRPC decorator to a service method. When a message matches the exchange and routing key, the method result is sent back via Direct Reply-To Queue.

```typescript
import { RabbitRPC } from '@golevelup/nestjs-rabbitmq';
import { Injectable } from '@nestjs/common';

@Injectable()
export class MessagingService {
  @RabbitRPC({
    exchange: 'exchange1',
    routingKey: 'rpc-route',
    queue: 'rpc-queue',
  })
  public async rpcHandler(msg: {}) {
    return {
      response: 42,
    };
  }
}
```

--------------------------------

### Expose Pub/Sub Handler with RabbitSubscribe

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Apply RabbitSubscribe decorator to a service method. When a message matches the exchange and routing key, the method is invoked with the message.

```typescript
import { RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
import { Injectable } from '@nestjs/common';

@Injectable()
export class MessagingService {
  @RabbitSubscribe({
    exchange: 'exchange1',
    routingKey: 'subscribe-route',
    queue: 'subscribe-queue',
  })
  public async pubSubHandler(msg: {}) {
    console.log(`Received message: ${JSON.stringify(msg)}`);
  }
}
```

--------------------------------

### AmqpConnection.publish method signature

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Signature of the publish method on AmqpConnection for fire-and-forget message publishing.

```typescript
public publish(
  exchange: string,
  routingKey: string,
  message: any,
  options?: amqplib.Options.Publish
)
```

--------------------------------

### Publish message to exchange

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Publish a message to an exchange with a routing key using the publish method.

```typescript
amqpConnection.publish('some-exchange', 'routing-key', { msg: 'hello world' });
```

--------------------------------

### request<T>

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Requests data from an RPC handler. This method sends a request and waits for a response, with an optional timeout.

```APIDOC
## request<T>

### Description
Requests data from an RPC handler. This method sends a request and waits for a response, with an optional timeout.

### Method
request

### Signature
```typescript
request<T>(options: {
  exchange: string;
  routingKey: string;
  payload: any;
  timeout?: number;
}): Promise<T>
```

### Parameters
- **exchange** (string) - Required - The exchange to send the request to.
- **routingKey** (string) - Required - The routing key for the request.
- **payload** (any) - Required - The request payload.
- **timeout** (number) - Optional - Timeout in milliseconds before the request fails if no response is received.

### Example
```typescript
const response = await amqpConnection.request<ExpectedReturnType>({
  exchange: 'exchange1',
  routingKey: 'rpc',
  payload: { request: 'val' },
  timeout: 10000,
});
```

### Type Inference
The generic parameter `T` specifies the expected return type. No runtime validation is performed.

### Interop
This RPC functionality is based on RabbitMQ's Direct Reply-To Queue. It may work with RPC servers in other languages/frameworks, but this has not been verified.
```

--------------------------------

### amqpConnection.request with correlationId

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Shows how to make an RPC request with a correlation ID and a request ID header for distributed transaction correlation.

```APIDOC
## amqpConnection.request

### Description
Makes an RPC request to an exchange with a routing key, including a correlation ID and a request ID header for distributed transaction correlation.

### Method
request

### Parameters
- **exchange** (string) - Required - The exchange to send the request to.
- **routingKey** (string) - Required - The routing key for the request.
- **correlationId** (string) - Required - The correlation ID for the transaction.
- **headers** (object) - Required - Headers to include, e.g., 'X-Request-ID'.
- **payload** (object) - Required - The request payload.

### Request Example
```typescript
const correlationId = randomUUID();
const response = await amqpConnection.request<ExpectedReturnType>({
  exchange: 'exchange1',
  routingKey: 'rpc',
  correlationId,
  headers: { 'X-Request-ID': randomUUID() },
  payload: {
    request: 'val',
  },
});
```

### Response
Returns a promise that resolves to the response of type `ExpectedReturnType`.
```

--------------------------------

### Identify RabbitMQ context with RABBIT_CONTEXT_TYPE_KEY

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Shows how to identify RabbitMQ contexts in a global interceptor using the context type key RABBIT_CONTEXT_TYPE_KEY. Global enhancers apply to all RabbitMQ handlers, so conditional logic is needed to skip them for 'rmq' contexts.

```typescript
import { RABBIT_CONTEXT_TYPE_KEY } from '@golevelup/nestjs-rabbitmq';
@Injectable()
class ExampleInterceptor implements NestInterceptor {
  intercept(context: ExecutionContext, next: CallHandler<any>) {
    const contextType = context.getType<
      'http' | typeof RABBIT_CONTEXT_TYPE_KEY
    >();

    // Do nothing if this is a RabbitMQ event
    if (contextType === RABBIT_CONTEXT_TYPE_KEY) {
      return next.handle();
    }

    // Execute custom interceptor logic for HTTP request/response
    return next.handle();
  }
}
```

--------------------------------

### Create a distributed transaction with correlationId and X-Request-ID

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Use this snippet to create a distributed transaction by correlating multiple requests with a correlationId and setting an X-Request-ID header for each sub-request.

```typescript
// To create a transaction in the distributed system,
// multiple request correlated by an correlationId
const correlationId = randomUUID();
const response = await amqpConnection.request<ExpectedReturnType>({
  exchange: 'exchange1',
  routingKey: 'rpc',
  correlationId,
  // Each request in the transaction has its own requestId
  headers: { 'X-Request-ID': randomUUID() },
  payload: {
    request: 'val',
  },
});
```

--------------------------------

### Handle message processing errors with Nack

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Shows how to handle message processing errors in a RabbitMQ subscriber. Returns `new Nack(true)` to requeue for retry (with a retry counter to avoid infinite loops) and `new Nack(false)` to send to the DLQ after exhausting retries.

```typescript
try {
  // ... process message ...
} catch (e) {
  console.error('Failed to process message', e);
  if (retryCount < 3) {
    // Requeue for retry — make sure you have a retry counter to avoid infinite loops
    return new Nack(true);
  }
  // Exhausted retries — nack without requeue (routes to DLQ if configured)
  return new Nack(false);
}
}
```

--------------------------------

### Handle queue creation errors with assertQueueErrorHandler

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Use this decorator to handle errors during queue creation (assertQueue operation). The default handler is `defaultAssertQueueErrorHandler` which rethrows the RabbitMq error. You can also use `forceDeleteAssertQueueErrorHandler` to delete and recreate the queue with provided queueOptions, or provide your own function that must return the name of the created queue.

```typescript
  @RabbitSubscribe({
    exchange: 'exchange1',
    routingKey: 'subscribe-route1',
    queue: 'subscribe-queue',
    assertQueueErrorHandler: myErrorHandler
  })
```

--------------------------------

### Identify RabbitMQ context with isRabbitContext

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Alternative way to identify RabbitMQ contexts using the isRabbitContext utility function. This is a drop-in replacement for the RABBIT_CONTEXT_TYPE_KEY check in global interceptors.

```typescript
import { isRabbitContext } from '@golevelup/nestjs-rabbitmq';

@Injectable()
class ExampleInterceptor implements NestInterceptor {
  intercept(context: ExecutionContext, next: CallHandler<any>) {
    // Do nothing if this is a RabbitMQ event
    if (isRabbitContext(context)) {
      return next.handle();
    }

    // Execute custom interceptor logic for HTTP request/response
    return next.handle();
  }
}
```

--------------------------------

### Custom deserializer and serializer for non-JSON messages

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Configure RabbitMQModule.forRoot with custom deserializer and serializer functions to handle messages in formats other than JSON. The deserializer receives a Buffer and ConsumeMessage, while the serializer converts a message object into a Buffer.

```typescript
@Module({
  imports: [
    RabbitMQModule.forRoot({
      // ...
      deserializer: (message: Buffer, msg: ConsumeMessage) => {
        const decodedMessage = myCustomDeserializer(
          msg.toString(),
          msg.properties.headers,
        );
        return decodedMessage;
      },
      serializer: (msg: any) => {
        const encodedMessage = myCustomSerializer(msg);
        return Buffer.from(encodedMessage);
      },
    }),
  ],
  // ...
})
export class RabbitExampleModule {}
```

--------------------------------

### Use Interceptor with RabbitRPC Handler

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Apply @UseInterceptors to a RabbitRPC handler. The controller must be imported as a provider in the module. The handler returns a message object.

```typescript
@RabbitRPC({
  routingKey: 'intercepted-rpc-2',
  exchange: 'exchange2',
  queue: 'intercepted-rpc-2',
})
@UseInterceptors(TransformInterceptor)
interceptedRpc() {
  return {
    message: 42,
  };
}
```

--------------------------------

### Prevent RabbitMQ connection in tests with undefined config

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Prevents connection to RabbitMQ during tests by passing `undefined` config to `RabbitMQModule.forRoot` when `NODE_ENV` is 'test'. This avoids setting up a broker locally or connecting to an external one.

```typescript
import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';

@Module({
  imports: [
    RabbitMQModule.forRoot(
      RabbitMQModule,
      /** Not sending config object makes the connection process to be ignored */
      process.env.NODE_ENV !== 'test'
        ? {
            exchanges: [
              {
                name: 'exchange1',
                type: 'topic',
              },
            ],
            uri: 'amqp://rabbitmq:rabbitmq@localhost:5672',
            connectionInitOptions: { wait: false },
          }
        : undefined,
    ),
  ],
})
export class RabbitExampleModule {}
```

--------------------------------

### Request data from RPC handler

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Request data from an RPC handler using the request method with a generic type and optional timeout.

```typescript
const response = await amqpConnection.request<ExpectedReturnType>({
  exchange: 'exchange1',
  routingKey: 'rpc',
  payload: {
    request: 'val',
  },
  timeout: 10000, // optional timeout for how long the request
  // should wait before failing if no response is received
});
```

--------------------------------

### Access original AMQP message with ConsumeMessage

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Accesses the original AMQP message by adding a second parameter of type `amqplib.ConsumeMessage` to the subscriber handler. The second argument is always the raw message, not a Channel, so use the `Nack` return value for acknowledgement.

```typescript
import { RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
import { Injectable } from '@nestjs/common';
import { ConsumeMessage } from 'amqplib';

@Injectable()
export class MessagingService {
  @RabbitSubscribe({
    exchange: 'exchange1',
    routingKey: 'subscribe-route',
    queue: 'subscribe-queue',
  })
  public async pubSubHandler(msg: {}, amqpMsg: ConsumeMessage) {
    console.log(`Correlation id: ${amqpMsg.properties.correlationId}`);
  }
}
```

--------------------------------

### Other Custom Errors

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Lists additional custom error classes available for granular error handling.

```APIDOC
## Custom Error Classes

### Description
Additional custom error classes for specific error handling.

### Errors
- **NullMessageError** - Thrown when a null message is received.
- **ChannelNotAvailableError** - Thrown when attempting to use a channel that is not available.
- **ConnectionNotAvailableError** - Thrown when attempting to use a connection that is not available.

All can be imported from `@golevelup/nestjs-rabbitmq`.
```

--------------------------------

### Inject AmqpConnection in Nest Controller

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Inject the AmqpConnection into a Nest controller or service via the constructor to access RabbitMQ interactions.

```typescript
@Controller()
export class AppController {
  constructor(private readonly amqpConnection: AmqpConnection) {}

  ...
}
```

--------------------------------

### Handle RpcTimeoutError with try-catch

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Use this snippet to catch and handle RpcTimeoutError when an RPC request times out. The error includes timeout duration, exchange, and routing key details.

```typescript
import { AmqpConnection, RpcTimeoutError } from '@golevelup/nestjs-rabbitmq';

try {
  const response = await amqpConnection.request<ExpectedReturnType>({
    exchange: 'exchange1',
    routingKey: 'rpc',
    payload: { request: 'val' },
    timeout: 5000,
  });
} catch (error) {
  if (error instanceof RpcTimeoutError) {
    console.error(`RPC timed out after ${error.timeout}ms`);
    console.error(
      `Exchange: ${error.exchange}, Routing Key: ${error.routingKey}`,
    );
    // Handle timeout specifically
  } else {
    // Handle other errors
    throw error;
  }
}
```

--------------------------------

### Publish message with persistent option

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Publish a message with the persistent option to mark it for broker durability.

```typescript
amqpConnection.publish(
  'some-exchange',
  'routing-key',
  { msg: 'hello world' },
  { persistent: true },
);
```

--------------------------------

### Nack a message without requeue in RabbitSubscribe handler

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Shows how to negatively acknowledge a message by returning a Nack instance from a RabbitSubscribe handler. By default, new Nack() nacks without requeueing; passing false explicitly sends the message to the dead-letter exchange if configured. Returning void/undefined auto-acks the message.

```typescript
import { RabbitSubscribe, Nack } from '@golevelup/nestjs-rabbitmq';
import { Injectable } from '@nestjs/common';

@Injectable()
export class MessagingService {
  @RabbitSubscribe({
    exchange: 'exchange1',
    routingKey: 'subscribe-route',
    queue: 'subscribe-queue',
  })
  public async pubSubHandler(msg: {}) {
    try {
      // ... process message ...
      // returning void/undefined here causes the library to auto-ack the message
    } catch (e) {
      // Log the error so it's not silently swallowed
      console.error('Failed to process message', e);
      // Nack without requeue — message goes to the dead-letter exchange (if configured)
      return new Nack(false);
    }
  }
}
```

--------------------------------

### RabbitSubscribe with custom errorHandler

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Configures a custom errorHandler for a RabbitSubscribe handler. When provided, it takes precedence over the errorBehavior enum. Use with rpcOptions for RPC.

```typescript
  @RabbitSubscribe({
    exchange: 'exchange1',
    routingKey: 'subscribe-route1',
    queue: 'subscribe-queue',
    errorHandler: myErrorHandler
  })
```

--------------------------------

### Use Pipe with RabbitRPC Handler

Source: https://golevelup.github.io/nestjs/modules/rabbitmq.html

Apply @UsePipes with ValidationPipe to a RabbitRPC handler. Includes errorBehavior and errorHandler options. The handler receives a message payload.

```typescript
@RabbitRPC({
  routingKey: 'intercepted-rpc-2',
  exchange: 'exchange2',
  queue: 'intercepted-rpc-2',
  errorBehavior: MessageHandlerErrorBehavior.ACK,
  errorHandler: ReplyErrorCallback,
})
@UsePipes(ValidationPipe)
interceptedRpc(@RabbitPayload()  message:messageDto) {
  return {
    message: 42,
  };
}
```
Tokens:
Raw