@anabranch/queue

Queue primitives with Task/Stream semantics for error-tolerant message processing.

Description

A queue abstraction that integrates with anabranch's Task and Stream types for composable error handling, concurrent processing, and automatic resource management.

Features

• Task/Stream Integration: Leverage Task's retry/timeout and Stream's error collection
• Multiple Adapters: In-memory implementation included, Redis/RabbitMQ/SQS coming soon
• Delayed Messages: Support for scheduled/delayed message delivery
• Dead Letter Queues: Automatic routing of failed messages after max attempts
• Batch Operations: Send multiple messages, acknowledge multiple at once

Installation

# JSR
jsr add @anabranch/queue

# Deno
deno add @anabranch/queue

Quick Start

import { createInMemory, Queue } from '@anabranch/queue'

const connector = createInMemory()
const queue = await Queue.connect(connector).run()

// Send a message
const id = await queue
  .send('notifications', { type: 'welcome', userId: 123 })
  .run()

// Process messages with error collection
const { successes, errors } = await queue
  .stream('notifications', { concurrency: 5 })
  .map(async (msg) => await sendNotification(msg.data))
  .tapErr((err) => logError(err))
  .collect()
  .then((results) => {
    const successes: typeof results = []
    const errors: typeof results = []
    for (const r of results) {
      if (r.type === 'success') successes.push(r)
      else errors.push(r)
    }
    return { successes, errors }
  })

API

Queue.send

Send a message to a queue with optional delay:

await queue.send('my-queue', { key: 'value' }, { delayMs: 30_000 }).run()

Queue.stream

Stream messages with concurrent processing:

const { successes, errors } = await queue
  .stream('orders', { count: 10, concurrency: 10 })
  .map(async (msg) => await processOrder(msg.data))
  .partition()

Queue.ack / Queue.nack

Acknowledge successful processing or negative acknowledge with requeue:

await queue.nack('orders', msg.id, { requeue: true, delay: 5_000 }).run()

// Or route to dead letter queue
await queue.nack('orders', msg.id, { deadLetter: true }).run()

Queue.sendBatch

Send multiple messages efficiently:

const ids = await queue
  .sendBatch('notifications', [
    { to: '[email protected]' },
    { to: '[email protected]' },
  ])
  .run()

Configuration

In-Memory Queue Options

const connector = createInMemory({
  maxBufferSize: 1000,
  queues: {
    orders: {
      maxAttempts: 3,
      deadLetterQueue: 'orders-failed',
    },
  },
})

Error Handling

All errors are typed for catchable handling:

• QueueConnectionFailed - Connection establishment failed
• QueueSendFailed - Send operation failed
• QueueReceiveFailed - Receive operation failed
• QueueAckFailed - Acknowledgment failed

try {
  await queue.send('my-queue', data).run()
} catch (error) {
  if (error instanceof QueueSendFailed) {
    console.error('Failed to send:', error.message)
  }
}

Related

• @anabranch/anabranch - Core Task/Stream primitives
• @anabranch/db - Database adapter pattern (inspiration for this package)