Fila

O Wippy fornece um sistema de filas para processamento assíncrono de mensagens com drivers e consumidores configuráveis.

Arquitetura

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

• Driver - Implementação de backend (memória, AMQP, SQS)
• Queue - Fila lógica vinculada a um driver
• Consumer - Conecta fila ao handler com configurações de concorrência
• Worker Pool - Processadores de mensagens concorrentes

Múltiplas filas podem compartilhar um driver. Múltiplos consumidores podem processar da mesma fila.

Tipos de Entradas

Configuração do Driver

Driver de Memória

Driver in-process para desenvolvimento e implantações de nó único. Sem dependências externas.

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

Driver AMQP

Para RabbitMQ e brokers compatíveis com 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

Bloco 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

Os campos inline cert/key/ca carregam conteúdo PEM; as variantes *_env são resolvidas através do registro env. As duas fontes são mutuamente exclusivas por campo. insecure_skip_verify desativa a verificação de certificado (apenas desenvolvimento).

Driver SQS

Para AWS SQS e endpoints compatíveis com SQS (LocalStack, ElasticMQ). Credenciais, região e outras configurações do AWS SDK vêm de um recurso config.aws compartilhado.

- 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

As filas são criadas automaticamente pelo driver no primeiro uso. Use headers com prefixo SQS (sqs.*) para endereçar atributos específicos do SQS na publicação; chaves neutras como correlation_id e content_type são traduzidas para atributos do sistema SQS quando possível.

Configuração de Fila

- 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

Opções do Driver

As chaves sob driver_options são agrupadas por nome do driver. Um driver lê apenas seu próprio sub-bag — as outras chaves ficam inativas, o que permite que uma única entrada de fila declare configurações para múltiplos drivers se necessário.

memory:

amqp:

Configuração do 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

Opções de consumidor amqp:

Pool de Workers

Workers executam como goroutines concorrentes:

concurrency: 3, prefetch: 10

1. Driver entrega até 10 mensagens para o buffer
2. 3 workers pegam do buffer concorrentemente
3. Conforme workers terminam, buffer reabastece
4. Contrapressão quando todos workers ocupados e buffer cheio

Função Handler

Handlers de consumidor recebem o corpo da mensagem decodificado como primeiro argumento. Use queue.message() para acessar metadados 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

Reconhecimento

O runtime faz settle automaticamente baseado no retorno do handler:

Chame msg:ack() ou msg:nack() explicitamente apenas para fazer settle antecipadamente. O settlement é de disparo único: vence a primeira chamada que chega.

Roteamento Dead-Letter

Quando dead_letter está configurado na fila, uma mensagem que é nack além de max_attempts é roteada para a DLQ com os headers x_dead_letter_reason e x_original_queue definidos pelo driver. Publicadores não devem definir nenhum header x_* — estes são reservados para registro da DLQ.

Publicando Mensagens

A partir de código Lua:

local queue = require("queue")

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

Veja Módulo Queue para API completa.

Encerramento Gracioso

Ao parar consumidor:

1. Para de aceitar novas entregas
2. Cancela contextos de workers
3. Aguarda mensagens em voo (com timeout)
4. Retorna erro se workers não terminarem a tempo

Veja Também

• Módulo Queue - Referência da API Lua
• Guia de Consumidores de Filas - Padrões de consumidor e pools de workers
• Supervisão - Gerenciamento de ciclo de vida do consumidor