nestjs-sqs

Tested with: AWS SQS and ElasticMQ.

Nestjs-sqs is a project to make SQS easier to use and control some required flows with NestJS. This module provides decorator-based message handling suited for simple use.

This library internally uses bbc/sqs-producer and bbc/sqs-consumer, and implements some more useful features on top of the basic functionality given by them.

Installation

npm i --save @ssut/nestjs-sqs @aws-sdk/client-sqs

Quick Start

Register module

Just register this module:

@Module({
  imports: [
    SqsModule.register({
      consumers: [
        {
          // Name is a unique identifier for this consumer instance
          name: "myConsumer1",
          // The actual SQS queue URL
          queueUrl: "https://sqs.region.amazonaws.com/account/queue-name",
          region: "us-east-1",
        },
      ],
      producers: [
        {
          // Name is a unique identifier for this producer instance
          name: "myProducer1",
          // The actual SQS queue URL
          queueUrl: "https://sqs.region.amazonaws.com/account/queue-name",
          region: "us-east-1",
        },
      ],
    }),
  ],
})
class AppModule {}

Quite often you might want to asynchronously pass module options instead of passing them beforehand. In such case, use registerAsync() method like many other Nest.js libraries.

• Use factory

SqsModule.registerAsync({
  useFactory: () => {
    return {
      consumers: [...],
      producers: [...],
    };
  },
});

• Use class

SqsModule.registerAsync({
  useClass: SqsConfigService,
});

• Use existing

SqsModule.registerAsync({
  imports: [ConfigModule],
  useExisting: ConfigService,
});

Decorate methods

You need to decorate methods in your NestJS providers in order to have them be automatically attached as event handlers for incoming SQS messages:

import { Message } from "@aws-sdk/client-sqs";

@Injectable()
export class AppMessageHandler {
  @SqsMessageHandler(/** name: */ "myConsumer1", /** batch: */ false)
  public async handleMessage(message: Message) {}

  @SqsConsumerEventHandler(
    /** name: */ "myConsumer1",
    /** eventName: */ "processing_error",
  )
  public onProcessingError(error: Error, message: Message) {
    // report errors here
  }
}

Produce messages

export class AppService {
  public constructor(
    private readonly sqsService: SqsService,
  ) { }

  public async dispatchSomething() {
    await this.sqsService.send(/** name: */ 'myProducer1', {
      id: 'id',
      body: { ... },
      groupId: 'groupId',
      deduplicationId: 'deduplicationId',
      messageAttributes: { ... },
      delaySeconds: 0,
    });
  }
}

Configuration

See here, and note that we have same configuration as bbc/sqs-consumer's. In most time you just need to specify both name and queueUrl at the minimum requirements.

License

This project is licensed under the terms of the MIT license.