SQS consumer using undici

Description

This module allows consuming SQS messages using @fgiova/mini-sqs-client thorough the aws-json protocol with "undici" as http agent .
The @fgiova/aws-signature module is used for signing requests to optimize performance.

Installation

npm install @fgiova/sqs-consumer

Usage

import {SQSConsumer} from '@fgiova/sqs-consumer'

const consumer = new SQSConsumer({
    queueARN: "arn:aws:sqs:eu-central-1:000000000000:test-queue-hooks",
    handler: async (message) => {
        console.log(message.Body);
    }
});

Handler cancellation on timeout

When handlerOptions.executionTimeout triggers, the consumer aborts the in-flight handler via an AbortSignal exposed through AsyncLocalStorage. The handler does not need to receive the signal as an argument — it is retrieved on demand using the getAbortSignal helper. Pass it to any cancellable async API (fetch, undici, timers, streams) to release resources promptly and avoid zombie handlers retaining memory after a timeout.

import { SQSConsumer, getAbortSignal } from '@fgiova/sqs-consumer'

const consumer = new SQSConsumer({
    queueARN: "arn:aws:sqs:eu-central-1:000000000000:test-queue-hooks",
    handler: async (message) => {
        const signal = getAbortSignal();
        const res = await fetch("https://api.example.com/work", { signal });
        return res.json();
    }
});

getAbortSignal() returns undefined when called outside a handler execution context. Handlers that ignore it keep working unchanged.

Options

Option	Type	Default	Description
queueARN	string		The ARN of the queue to consume
handler	(message: Message) => Promise		The handler function to be called for each message
logger	Logger	console	The logger to be used
autoStart	boolean	true	Whether to start the consumer automatically
handlerOptions	HandlerOptions		The options for the handler
clientOptions	ClientOptions		The options for the client
consumerOptions	ConsumerOptions		The options for the consumer
hooks	HooksOptions		The hooks to be called on specific events

HandlerOptions

Option	Type	Default	Description
deleteMessage	boolean	true	Whether to delete the message after handling (if handler execute without any error)
extendVisibilityTimeout	boolean	true	Whether to extend the visibility timeout during message handling
executionTimeout	number	30000	The timeout for the handler execution in ms
parallelExecution	boolean	true	If true execute handler in parallel for each batch of messages received, otherwise execute consecutively

ClientOptions

Option	Type	Default	Description
sqsClient	MiniSQSClient		The MiniSQSClient client to be used. If not provided, a new client will be created using the queueARN.
endpoint	string		The endpoint to be used for the client. If not provided, the endpoint will be inferred from the queueARN.
undiciOptions	Pool.Options		The options for the undici client.
signer	Signer / SignerOptions		The signer to be used for signing requests. If not provided, a new singleton signer will be created.
destroySigner	boolean	true	Whether to destroy the signer when the consumer is destroyed.

ConsumerOptions

Option	Type	Default	Description
visibilityTimeout	number	30	The visibility timeout for the messages in seconds
waitTimeSeconds	number	20	The wait time for the receiveMessage call in seconds
itemsPerRequest	number	10	The maximum number of messages to be received at once
messageAttributeNames	string[]	[]	The message attribute names to be included in the response
attributeNames	string[]	[]	The attribute names to be included in the response

Hooks

Option	Type	Description
onPoll	(messages: Message[]) => Promise<Message[]>	Called when the consumer polls for messages
onMessage	(message: Message) => Promise	Called when the consumer receives a message
onHandlerSuccess	(message: Message) => Promise	Called when the consumer handles a message successfully
onHandlerTimeout	(message: Message) => Promise	Called when the consumer handler execution exceed executionTimeout
onHandlerError	(message: Message, error: Error) => Promise	Called when the consumer handler execution throws an error
onSuccess	(message: Message) => Promise	Called when the consumer handler execution finishes successfully
onError	(hook: HookName, message: Message, error: Error) => Promise	Called when the consumer handler execution throws an uncaught error
onSQSError	(error: Error, message?: Message) => Promise	Called when the consumer receives an error from the SQS service

API

SQSConsumer(options: SQSConsumerOptions)
SQSConsumer.start(): Promise<void>
SQSConsumer.stop(destroyConsumer = false): Promise<void>
SQSConsumer.addHook<H extends HookName>(hookName: H, value: HookCallback<H>): this
getAbortSignal(): AbortSignal | undefined

License

Licensed under MIT.