Cola

Wippy proporciona un sistema de colas para procesamiento asíncrono de mensajes con drivers y consumidores configurables.

Arquitectura

flowchart LR
    P[Publicador] --> D[Driver]
    D --> Q[Cola]
    Q --> C[Consumidor]
    C --> W[Pool de Workers]
    W --> F[Función]

• Driver - Implementación de backend (memory, AMQP, SQS)
• Cola - Cola lógica vinculada a un driver
• Consumidor - Conecta cola a handler con configuración de concurrencia
• Pool de Workers - Procesadores de mensajes concurrentes

Múltiples colas pueden compartir un driver. Múltiples consumidores pueden procesar de la misma cola.

Tipos de Entrada

Configuración de Driver

Driver de Memoria

Driver in-process para desarrollo y despliegues de un solo nodo. Sin dependencias externas.

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

Driver AMQP

Para RabbitMQ y brokers compatibles con AMQP 0-9-1.

- 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

Bloque TLS:

  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

Los campos inline cert/key/ca contienen contenido PEM; las variantes *_env se resuelven a través del registro env. Las dos fuentes son mutuamente excluyentes por campo. insecure_skip_verify desactiva la verificación de certificado (solo desarrollo).

Driver SQS

Para AWS SQS y endpoints compatibles con SQS (LocalStack, ElasticMQ). Las credenciales, región y otras configuraciones del AWS SDK provienen de un recurso config.aws compartido.

- 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

Las colas son creadas automáticamente por el driver en el primer uso. Use headers con prefijo SQS (sqs.*) para direccionar atributos específicos de SQS al publicar; las claves neutrales como correlation_id y content_type se traducen a atributos del sistema SQS cuando es posible.

Configuración de Cola

- 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

Opciones de Driver

Las claves bajo driver_options están agrupadas por nombre de driver. Un driver lee solo su propio sub-bag — las otras claves quedan inactivas, lo que permite que una sola entrada de cola declare configuraciones para múltiples drivers si es necesario.

memory:

amqp:

Configuración de Consumidor

- 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

Opciones de consumidor amqp:

Pool de Workers

Los workers se ejecutan como goroutines concurrentes:

concurrency: 3, prefetch: 10

1. El driver entrega hasta 10 mensajes al buffer
2. 3 workers extraen del buffer concurrentemente
3. A medida que los workers terminan, el buffer se rellena
4. Backpressure cuando todos los workers están ocupados y el buffer lleno

Función Handler

Los handlers de consumidor reciben el cuerpo decodificado del mensaje como primer argumento. Use queue.message() para acceder a metadatos de entrega (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 or DLQ
    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

Reconocimiento

El runtime hace settle automáticamente según el retorno del handler:

Llame msg:ack() o msg:nack() explícitamente solo para hacer settle anticipado. El settlement es de un solo disparo: gana la primera llamada que llega.

Enrutamiento Dead-Letter

Cuando dead_letter está configurado en la cola, un mensaje que es nack más allá de max_attempts es enrutado a la DLQ con los headers x_dead_letter_reason y x_original_queue establecidos por el driver. Los publicadores no deben establecer ningún header x_* — estos están reservados para el registro de DLQ.

Publicando Mensajes

Desde código Lua:

local queue = require("queue")

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

Ver Módulo Queue para la API completa.

Apagado Graceful

Al detener el consumidor:

1. Dejar de aceptar nuevas entregas
2. Cancelar contextos de workers
3. Esperar mensajes en vuelo (con timeout)
4. Retornar error si los workers no terminan a tiempo

Ver También

• Módulo Queue - Referencia de API Lua
• Guía de Consumidores de Cola - Patrones de consumidor y pools de workers
• Supervisión - Gestión del ciclo de vida del consumidor