Queue

Wippy provides a queue system for asynchronous message processing with configurable drivers and consumers.

Architecture

flowchart LR
    P[Publisher] --> D[Driver]
    D --> Q[Queue]
    Q --> C[Consumer]
    C --> W[Worker Pool]
    W --> F[Function]

• Driver - Backend implementation (memory, AMQP, SQS)
• Queue - Logical queue bound to a driver
• Consumer - Connects queue to handler with concurrency settings
• Worker Pool - Concurrent message processors

Multiple queues can share a driver. Multiple consumers can process from the same queue.

Entry Kinds

Driver Configuration

Memory Driver

In-process driver for development and single-node deployments. No external dependencies.

- name: memory_driver
  kind: queue.driver.memory
  lifecycle:
    auto_start: true

AMQP Driver

For RabbitMQ and AMQP 0-9-1 compatible brokers.

- name: amqp_driver
  kind: queue.driver.amqp
  url: "amqp://guest:guest@localhost:5672/"
  vhost: "/"
  connection_name: "wippy-service"
  heartbeat: "10s"
  connection_timeout: "30s"
  reconnect_delay: "1s"
  reconnect_max_delay: "30s"
  default_message_ttl: "1h"
  default_queue_expiry: "24h"
  prefetch_count: 10
  lifecycle:
    auto_start: true

TLS block:

  tls:
    enabled: true
    server_name: "rabbit.example.com"
    cert_env: "AMQP_CLIENT_CERT"
    key_env: "AMQP_CLIENT_KEY"
    ca_env: "AMQP_CA_CERT"
    insecure_skip_verify: false

Inline cert/key/ca fields carry PEM content; *_env variants resolve through the env registry. The two sources are mutually exclusive per field. insecure_skip_verify disables certificate verification (development only).

SQS Driver

For AWS SQS and SQS-compatible endpoints (LocalStack, ElasticMQ). Credentials, region, and other AWS SDK settings come from a shared config.aws resource.

- name: aws_config
  kind: config.aws
  region: us-east-1
  access_key_id_env: app:AWS_ACCESS_KEY_ID
  secret_access_key_env: app:AWS_SECRET_ACCESS_KEY

- name: sqs_driver
  kind: queue.driver.sqs
  config: app:aws_config
  endpoint: "http://localhost:9324"
  message_retention_period: 345600
  default_delay_seconds: 0
  lifecycle:
    auto_start: true

Queues are auto-created by the driver on first use. Use SQS-prefixed headers to address SQS-specific fields on publish: sqs.delay_seconds, sqs.message_group_id, and sqs.message_deduplication_id map to typed SQS message fields. All other headers (neutral keys like correlation_id and content_type, plus any sqs.message_attributes.* keys) are carried verbatim as SQS message attributes.

Queue Configuration

- name: tasks
  kind: queue.queue
  driver: app.queue:memory_driver
  codec: json/plain
  queue_name: "app_tasks"
  driver_options:
    memory:
      max_length: 500
  dead_letter:
    queue: app.queue:tasks_dlq
    max_attempts: 5

Driver Options

Keys under driver_options are scoped by driver name. A driver reads only its own sub-bag — other keys are dormant, which lets a single queue entry declare settings for multiple drivers if needed.

memory:

amqp:

Codecs

The codec selects how a message body is serialized before it is handed to the broker. It is a payload format string and defaults to json/plain:

The AMQP driver sets a matching content-type (application/json or application/msgpack) on published messages. An unknown codec fails when the queue is declared, not at publish time.

Consumer Configuration

- name: task_consumer
  kind: queue.consumer
  queue: app.queue:tasks
  func: app.queue:task_handler
  concurrency: 4
  prefetch: 20
  auto_ack: false
  driver_options:
    amqp:
      consumer_tag: "worker-1"
      exclusive: false
  lifecycle:
    auto_start: true
    depends_on:
      - app.queue:tasks

amqp consumer options:

Worker Pool

Workers run as concurrent goroutines:

concurrency: 3, prefetch: 10

1. Driver delivers up to 10 messages to buffer
2. 3 workers pull from buffer concurrently
3. As workers finish, buffer refills
4. Backpressure when all workers busy and buffer full

Handler Function

Consumer handlers receive the decoded message body as the first argument. Use queue.message() to access delivery metadata (id, headers).

local queue = require("queue")
local logger = require("logger")

local function main(body)
    local msg = queue.message()
    logger:info("processing", {
        id = msg:id(),
        correlation_id = msg:header("correlation_id")
    })

    local ok, err = process_task(body)
    if err then
        return false  -- nack: redelivery per driver
    end
    return true       -- ack: remove from queue
end

return { main = main }

- name: task_handler
  kind: function.lua
  source: file://task_handler.lua
  method: main
  modules:
    - queue
    - logger

Acknowledgment

The runtime auto-settles based on the handler return:

Call msg:ack() or msg:nack() explicitly only to settle early. Settlement is single-shot: whichever call lands first wins.

Dead-Letter Routing

Dead-letter routing is not yet implemented. The dead_letter block (see Queue Configuration) is accepted in config, but no built-in driver currently counts attempts, routes nacked messages to the configured DLQ, or sets x_dead_letter_* headers. A nacked message is redelivered per the driver's own policy. The x_* header namespace is reserved for future DLQ bookkeeping, so publishers should avoid setting x_* headers.

Publishing Messages

From Lua code:

local queue = require("queue")

queue.publish("app.queue:tasks", {
    id = "task-123",
    action = "process",
    data = payload
})

See Queue Module for full API.

Graceful Shutdown

On consumer stop:

1. Stop accepting new deliveries
2. Cancel worker contexts
3. Wait for in-flight messages (with timeout)
4. Return error if workers don't finish in time

See Also

• Queue Module - Lua API reference
• Queue Consumers Guide - Consumer patterns and worker pools
• Supervision - Consumer lifecycle management