sqs-consumer

To shut down the consumer gracefully, call consumer.stop().

• Default behavior: stop() prevents new polls but emits stopped immediately, even if in-flight messages are still being processed.
• Graceful shutdown: Set pollingCompleteWaitTimeMs to a value (in milliseconds) to allow the consumer to wait for the last poll and current message handlers to finish.
• Abort option: Passing { abort: true } to stop() will cancel the shared AbortController, halting heartbeat extensions and preventing acknowledgments/deletions from finishing. For graceful shutdowns, keep abort: false (the default).

During the wait period, the consumer emits waiting_for_polling_to_complete every second. If the timeout is reached, waiting_for_polling_to_complete_timeout_exceeded is emitted before stopped.

const consumer = Consumer.create({
  queueUrl: "https://sqs.eu-west-1.amazonaws.com/account-id/queue-name",
  handleMessage: async (message) => {
    await doWork(message);
    return message;
  },
  pollingCompleteWaitTimeMs: 10_000, // This will allow up to 10s for the last poll + handler
});

const shutdown = (signal) => {
  console.log(`Received ${signal}, waiting for in-flight work...`);
  consumer.stop();

  consumer.once("waiting_for_polling_to_complete", () => {
    console.log("Still processing in-flight messages...");
  });

  consumer.once("waiting_for_polling_to_complete_timeout_exceeded", () => {
    console.warn("Graceful shutdown timed out, continuing shutdown anyway.");
  });

  consumer.once("stopped", () => {
    console.log("Consumer stopped cleanly");
    process.exit(0);
  });
};

process.once("SIGINT", shutdown);
process.once("SIGTERM", shutdown);

consumer.start();

The consumer uses the return value of your handler to decide whether to delete a message from SQS (acknowledgement).

For handleMessage (Single Message):

• Return the message object (or an object with the same MessageId): The message is considered successfully processed and will be deleted from SQS.
• Return undefined or null: The message is NOT deleted. If strictReturn is enabled in options, returning null will throw an error.
• Return a different object: If the returned object has the same MessageId, it is deleted.
• alwaysAcknowledge: true: If this option is set, the consumer will delete the message regardless of what the handler returns.

For handleMessageBatch (Batch):

• Return an array of Message objects: Only the messages included in this array will be deleted from SQS.
• Return undefined or null: No messages in the batch will be deleted.
• alwaysAcknowledge: true: All messages in the batch will be deleted regardless of the return value.

The ConsumerOptions interface defines the configuration for an SQS consumer. Key configuration areas include:

Message Handling

• handleMessage: An async function called for each message.
  • To acknowledge/delete: Return the original Message or a processed Message.
  • To NOT acknowledge: Return undefined.
  • To selectively acknowledge: Return an object containing the MessageId you want to acknowledge.
• handleMessageBatch: An async function called for a batch of messages. If set, it overrides handleMessage.
  • To acknowledge all: Return the original or processed array of Messages.
  • To NOT acknowledge any: Return undefined or an empty array [].
  • To selectively acknowledge: Return an array containing only the messages you want to acknowledge.
• alwaysAcknowledge: If true, all messages are acknowledged regardless of the handler's return value. (Default: false)
• shouldDeleteMessages: If false, the consumer will not delete messages from SQS. (Default: true)
• strictReturn: If true, handlers returning null will throw an error instead of being treated as "do not acknowledge". (Default: false)

Polling and Timing

• batchSize: Number of messages to request per poll (Max 10). (Default: 1)
• visibilityTimeout: Duration (seconds) messages are hidden after retrieval. (Default: 30 via SQS, but configurable)
• waitTimeSeconds: Duration (seconds) to wait for a message to arrive (Long Polling). (Default: 20)
• handleMessageTimeout: Time (ms) to wait for handleMessage to process before timing out. Emits timeout_error.
• pollingWaitTimeMs: Duration (ms) to wait before repolling. (Default: 0)

AWS and Client Configuration

• sqs: An optional SQSClient instance.
• region: The AWS region. (Default: process.env.AWS_REGION || 'eu-west-1')
• useQueueUrlAsEndpoint: If false, uses the client's resolved endpoint instead of the queueUrl. (Default: true)

Error and Visibility Management

• terminateVisibilityTimeout: Sets visibility timeout to 0 after a processing_error. Can be true, a number, or a function (messages: Message[]) => number for exponential backoff.
• heartbeatInterval: Interval (seconds) between requests to extend visibility timeout. Must be less than visibilityTimeout.