Package @occupop/lib-queue

This package is intended to be used with @occupop/lib-container providing queue consumer/publisher through AWS SQS/SNS clients.

This implementation refers to:

• https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/client/sqs
• https://docs.aws.amazon.com/sdk-for-javascript/v3/developer-guide/javascript_sqs_code_examples.html
• https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/client/sns
• https://docs.aws.amazon.com/sdk-for-javascript/v3/developer-guide/javascript_sns_code_examples.html

NPM registry token required!

Install

# Yarn
yarn add @occupop/lib-queue

# NPM
npm install @occupop/lib-queue

# Bun
bun add @occupop/lib-queue

Compatibility

TypeScript 5.x and 6.x: peerDependencies.typescript is ^5.0.0 || ^6.0.0 since 4.1.0 (4.0.0 and earlier declare ^5.0.0).

Env Setup

ENV variables required

AWS_ENDPOINT=http://localstack:4566
AWS_REGION=eu-west-1
AWS_ACCESS_KEY_ID=...
AWS_SECRET_ACCESS_KEY=...

AWS_SQS_QUEUE_URL=http://localstack:4566/000000000000/queue-name-here
AWS_SNS_TOPIC_ARN=arn:aws:sns:eu-west-1:000000000000:topic-name-here

Register

import { makeQueueService, makeSqsClient, makeSnsClient } from '@occupop/lib-queue'
import { eventContainer } from '.event-container'
//... all other imports omited... 

export const container = createTypedContainer({
    // ...
    // queue service:
    queueService: { asFunction: makeQueueService, singleton: true },
    // optional:
    eventContainer: { asValue: eventContainer },
    sqsClient: { asFunction: makeSqsClient, singleton: true },
    snsClient: { asFunction: makeSnsClient, singleton: true },
    // ...
})

• eventContainer can be provided direct on consumer call.
• sqsClient/snsClient is to be registered only if you need some kind of custom implementation.

Usage (worker)

// worker.ts
import { container } from '.container'
import { eventContainer } from '.event-container'

const { mongoClient, queueService } = container.cradle

await mongoClient.connect()
console.log('Mongo connected')

// providing eventContainer...
await queueService.consume({ eventContainer })
console.log('Queue consuming')

// ... or without, if you register on main container!
// await queueService.consume()
// console.log('Queue consuming')

Message failures

consume() handles each message of a received batch on its own:

• A handler that throws, or a body that is not JSON, is empty or missing, or parses to a falsy JSON value (null, 0, false, ""), is logged as Message handle failed and is not deleted. The queue redelivers it after its visibility timeout, and its redrive policy dead-letters it at maxReceiveCount.
• The rest of the batch is unaffected: the other messages are handled and deleted as usual. Once every message of the batch has been handled, the loop polls again after delaySeconds.
• A failed receive is logged as sqsClient error: failed to receiveMessage, and the loop waits failedDelaySeconds before it polls again.
• A parsed body whose event has no registered handler is deleted as ignored.

Usage (eventContainer simple implementation)

// event-container.ts
import type { QueueEventContainer, QueueEvent } from '@occupop/lib-queue'

export const eventContainer = {
    Event_name_here: (event: QueueEvent) => {
        // ... handler implementation
    },
    Another_event_there: (event: QueueEvent) => {
        // ... handler implementation
    }
}

Usage (eventContainer with dependency injection, recomended)

// event-container.ts
import { container } from '.container'
import type { QueueEventContainer, QueueEvent } from '@occupop/lib-queue'
import { createTypedContainerScope } from '@occupop/lib-container'

export const eventContainer = createTypedContainerScope(container, {
    Event_name_here: { asFunction: makeEventNameHandler },
    Another_event_there: { asFunction: makeAnotherEventHandler },
    Another_event_alike: { asFunction: makeAnotherEventHandler },
}).cradle

Sample event handler

// event-name-handler.ts
import type { Deps } from '.container'
import { QueueEvent } from "@occupop/lib-queue";

interface CustomEventName extends QueueEvent {
    customAttribute: string
    randomEntity: {
        uuid: string
        someOtherAttribute: string
        maybeAnOptionalOne?: boolean
    }
}

export const makeEventNameHandler = ({ fooService, barService }: Deps) => {
    (event: CustomEventName) => {
        // implementation that depends on fooService and barService...
    }
}