Fila

O sistema de filas conecta publicadores assíncronos, drivers, filas, consumidores e funções handler.

Esta página é uma referência de configuração e comportamento. Os blocos YAML são fragmentos de uma lista de entradas, salvo quando mostram um documento completo; exemplos de drivers externos pressupõem que o broker ou serviço compatível com AWS já exista.

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

Configure TLS em tls:

  tls:
    enabled: true
    server_name: "rabbit.example.com"
    cert: ${env:app.env:amqp_cert}
    key:  ${env:app.env:amqp_key}
    ca:   ${env:app.env:amqp_ca}
    insecure_skip_verify: false

cert/key/ca carregam conteúdo PEM — inline, via file://, ou via um placeholder ${env:NAME} resolvido através do registro env. insecure_skip_verify desativa a verificação de certificado (apenas desenvolvimento). As diretivas legadas cert_env/key_env/ca_env resolvem da mesma forma, mas estão obsoletas; prefira ${env:NAME}.

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: 86400
  default_delay_seconds: 0
  lifecycle:
    auto_start: true

As filas são criadas automaticamente pelo driver no primeiro uso. Use headers com prefixo SQS para endereçar campos específicos do SQS na publicação: sqs.delay_seconds, sqs.message_group_id e sqs.message_deduplication_id mapeiam para campos tipados de mensagem SQS. Todos os demais headers (chaves neutras como correlation_id e content_type, mais quaisquer chaves sqs.message_attributes.*) são carregados verbatim como atributos de mensagem SQS.

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:

Codecs

O codec seleciona como o corpo de uma mensagem é serializado antes de ser entregue ao broker. É uma string de formato de payload e usa por padrão json/plain:

O driver AMQP define um content-type correspondente (application/json ou application/msgpack) nas mensagens publicadas. Um codec desconhecido falha quando a fila é declarada, não no momento da publicação.

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
    requires:
      - app.queue:tasks

Opções de consumidor amqp:

Pool de Workers

Workers executam como goroutines concorrentes:

concurrency: 3, prefetch: 10

1. Driver delivers up to 10 messages to the shared buffer
2. 3 workers pull from the buffer and can each hold an active delivery
3. As workers finish, buffer refills
4. Backpressure when all workers busy and buffer full

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, msg_err = queue.message()
    if msg_err then return nil, msg_err end
    local message_id, id_err = msg:id()
    if id_err then return nil, id_err end
    local correlation_id, header_err = msg:header("correlation_id")
    if header_err then return nil, header_err end

    logger:info("processing", {
        id = message_id,
        correlation_id = correlation_id
    })

    local ok, err = process_task(body)
    if err then
        return nil, err  -- 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

Reconhecimento

Salvo quando o handler faz settle explícito, o consumidor decide pelo resultado da invocação da função:

Valores de retorno comuns, inclusive false, não selecionam o reconhecimento. Chame msg:ack() ou msg:nack() para fazer settle explícito. O settlement ocorre uma única vez: vence a primeira chamada.

Roteamento Dead-Letter

O roteamento dead-letter ainda não está implementado. O bloco dead_letter (veja Configuração de Fila) é aceito na configuração, mas nenhum driver integrado atualmente conta tentativas, roteia mensagens nack para a DLQ configurada ou define headers x_dead_letter_*. Uma mensagem nack é reentregue conforme a política do próprio driver. O namespace de headers x_* é reservado para registro futuro da DLQ, portanto publicadores devem evitar definir headers x_*.

Publicando Mensagens

A partir de código Lua:

local queue = require("queue")

local published, publish_err = queue.publish("app.queue:tasks", {
    id = "task-123",
    action = "process",
    data = payload
})
if publish_err then return nil, publish_err end
return published

Consulte o Módulo Queue para a API de publicação e mensagens em Lua.

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