Publicar no MQTT a partir do AWS Lambda e funções serverless

Você pode publicar em um broker MQTT a partir do AWS Lambda (ou de qualquer função serverless) com uma biblioteca cliente MQTT comum — o problema é o ciclo de vida. Uma função vive algumas centenas de milissegundos, é congelada entre invocações e pode rodar em dezenas de cópias paralelas. Conecte sob demanda, reaproveite o cliente entre invocações a quente, espere a confirmação do publish antes de retornar e dê a cada ambiente de execução seu próprio client ID. Se você só publica, uma ponte HTTP pode ser ainda mais simples.

Este tutorial mostra um handler Python pronto para produção (paho-mqtt 2.x), a mesma ideia em Node.js, a alternativa HTTP e como testar tudo.

O ciclo de vida serverless versus MQTT

O MQTT pressupõe conexões TCP de longa duração com pings de keep-alive. O Lambda pressupõe execuções curtas e sem estado. O descompasso aparece em quatro pontos:

O congelamento é o ponto sutil. Se um ambiente ficar ocioso por mais de cerca de 1,5× o intervalo de keep-alive, o broker fecha a conexão. Quando o ambiente descongela, o cliente ainda acha que está conectado até tentar escrever. Um cliente em cache ingênuo vai falhar ou travar na primeira invocação depois de um período parado.

Estratégia 1: conectar a cada invocação

O padrão correto mais simples: conectar, publicar, esperar o PUBACK, desconectar. Nada sobrevive entre chamadas, então nada fica obsoleto.

• Prós: nenhum bug de conexão obsoleta, nenhuma colisão de client ID se você usar um ID aleatório, fácil de entender.
• Contras: cada invocação paga o handshake de conexão — bem mais com TLS e a partir de uma região distante — e uma rajada de invocações vira uma rajada de novas conexões no broker.

É uma boa escolha para gatilhos de baixo volume: um job noturno, uma notificação de upload no S3, um webhook ocasional.

Estratégia 2: reaproveitar a conexão entre invocações a quente

Para qualquer coisa frequente, crie o cliente no escopo global para que as invocações a quente pulem o handshake, e desconfie da conexão em cache depois de um período ocioso.

import json, os, threading, time, uuid
import paho.mqtt.client as mqtt

HOST = os.environ.get("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.environ.get("MQTT_PORT", "8883"))
KEEPALIVE = 60
MAX_IDLE = KEEPALIVE            # depois desse intervalo, assuma que o broker nos derrubou

_client = None
_connected = threading.Event()
_last_used = 0.0

def _on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"connect refused: {reason_code}")
    else:
        _connected.set()

def _on_disconnect(client, userdata, flags, reason_code, properties):
    _connected.clear()

def _new_client():
    c = mqtt.Client(
        mqtt.CallbackAPIVersion.VERSION2,
        client_id=f"lambda-{uuid.uuid4().hex[:12]}",   # único por ambiente
        protocol=mqtt.MQTTv5,
    )
    c.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
    c.tls_set()                                         # bundle de CAs do sistema
    c.on_connect, c.on_disconnect = _on_connect, _on_disconnect
    c.connect_async(HOST, PORT, keepalive=KEEPALIVE)
    c.loop_start()
    return c

def _get_client(timeout=5.0):
    global _client
    if _client is not None and time.time() - _last_used > MAX_IDLE:
        _client.disconnect()                            # obsoleta após um congelamento
        _client.loop_stop()
        _connected.clear()
        _client = None
    if _client is None:
        _client = _new_client()
    if not _connected.wait(timeout):
        raise TimeoutError("MQTT connect timed out")
    return _client

def handler(event, context):
    global _last_used
    client = _get_client()
    payload = json.dumps({"source": "lambda", "event": event, "ts": time.time()})
    info = client.publish("testmqtt/lambda/events", payload, qos=1)
    info.wait_for_publish(timeout=5)
    if not info.is_published():
        raise RuntimeError("PUBACK not received")       # deixe o Lambda tentar de novo
    _last_used = time.time()
    return {"statusCode": 200, "body": json.dumps({"mid": info.mid})}

O que cada parte faz:

• connect_async + loop_start — o loop de rede roda em uma thread em segundo plano, que cuida de CONNACK, PUBACK e reconexões enquanto a função está rodando.
• MAX_IDLE — se o ambiente ficou ocioso mais que o keep-alive, o código descarta o cliente e reconecta em vez de escrever em um socket morto.
• wait_for_publish — nunca retorne antes de chegar a confirmação do QoS 1. Depois que o handler retorna, o ambiente pode ser congelado com a mensagem ainda no buffer do cliente.
• Lançar erro em caso de falha — em gatilhos assíncronos, o Lambda tenta de novo as invocações que falharam, que é exatamente o que você quer para um publish perdido.
• Client ID aleatório — dois ambientes concorrentes com o mesmo ID ficariam se derrubando (client ID duplicado).

Empacote com pip install paho-mqtt -t package/ e compacte a pasta junto com o handler, ou use uma Lambda layer. Defina o timeout da função bem acima do timeout de conexão (10–15 s é um bom ponto de partida) e guarde as credenciais em variáveis de ambiente alimentadas pelo Secrets Manager ou pelo Parameter Store.

O mesmo padrão em Node.js

Com o MQTT.js 5, o cliente em cache é uma variável no nível do módulo. A biblioteca reconecta sozinha, mas depois de um congelamento ainda vale verificar a conexão antes de publicar.

import mqtt from 'mqtt';
import { randomUUID } from 'node:crypto';

let client;
let lastUsed = 0;

async function getClient() {
  if (client && Date.now() - lastUsed > 60_000) { await client.endAsync(true); client = undefined; }
  if (!client) {
    client = await mqtt.connectAsync('mqtts://broker.connectmq.com:8883', {
      clientId: `lambda-${randomUUID().slice(0, 12)}`,
      username: process.env.MQTT_USER,
      password: process.env.MQTT_PASS,
      keepalive: 60,
      connectTimeout: 5000,
    });
  }
  return client;
}

export const handler = async (event) => {
  const c = await getClient();
  await c.publishAsync('testmqtt/lambda/events', JSON.stringify({ source: 'lambda', event }), { qos: 1 });
  lastUsed = Date.now();
  return { statusCode: 200 };
};

publishAsync com QoS 1 só resolve depois do PUBACK, então retornar após o await é seguro.

Estratégia 3: publicar via HTTP

Se suas funções só publicam e não precisam de semântica MQTT no cliente, uma requisição HTTP combina naturalmente com serverless: sem estado de conexão, sem client IDs, sem problemas de congelamento.

• APIs HTTP do broker. Alguns brokers expõem um endpoint REST que publica uma mensagem por você — por exemplo a API HTTP do EMQX e o endpoint HTTPS do data plane do AWS IoT Core. Veja a documentação do seu broker para autenticação e formato de payload.
• Sua própria ponte. Um pequeno serviço sempre ligado que mantém uma conexão MQTT e aceita requisições POST /publish autenticadas. Ele centraliza as credenciais e transforma mil conexões efêmeras de Lambda em uma única persistente.

A contrapartida: você perde a QoS de ponta a ponta — o HTTP 200 diz que a ponte aceitou a mensagem, não que um subscriber a recebeu — e ganha mais um componente para operar. Para uma comparação mais ampla dos dois protocolos, veja MQTT vs HTTP.

E para assinar?

Uma função Lambda não pode ser um subscriber MQTT de longa duração: ela só roda enquanto trata uma invocação. Inverta a direção — deixe algo sempre ligado assinar e invocar a função:

• Uma regra ou integração do broker que encaminha as mensagens correspondentes para um endpoint HTTP (API Gateway ou uma function URL do Lambda). Veja MQTT para webhook para o padrão geral.
• Um pequeno serviço assinante que envia as mensagens para o SQS, que por sua vez dispara o Lambda em lotes.

Erros comuns

• Task timed out after 3.00 seconds — o timeout padrão do Lambda é menor que uma conexão TLS a frio a partir de outra região. Aumente-o e confira se a porta do broker está acessível.
• O connect trava dentro de uma VPC — funções em VPC precisam de um NAT gateway para sair para a internet. Veja timeout de conexão MQTT.
• Not authorized / reason code 0x87 — credenciais erradas ou uma ACL que não libera o topic; a referência de reason codes traz o significado exato.
• Mensagens somem depois que a função retornou — você retornou antes do PUBACK. Sempre espere o publish terminar.

Testando

1. Abra o cliente MQTT online, conecte em broker.connectmq.com e assine testmqtt/lambda/#.
2. Invoque a função pelo console ou com aws lambda invoke --function-name mqtt-publisher --payload '{"hello":"world"}' --cli-binary-format raw-in-base64-out out.json.
3. A mensagem deve aparecer no navegador em menos de um segundo. Invoque de novo logo em seguida: a invocação a quente deve ser visivelmente mais rápida na métrica de duração da função.
4. Espere alguns minutos, invoque outra vez e confirme que o caminho de conexão obsoleta reconecta de forma limpa em vez de dar timeout.

Veja as mensagens do seu Lambda chegando

O endpoint de teste compartilhado tem limite de taxa e não guarda sessões, o que basta para essa verificação; para testes de carga com muitas invocações paralelas, aponte a função para a sua própria instância privada de broker.

Serverless e MQTT funcionam bem juntos quando você respeita o ciclo de vida: conecte sob demanda, desconfie de conexões depois de um congelamento, espere a confirmação e dê a cada ambiente sua própria identidade. Quando você só precisa publicar em escala sem esperar resposta, uma ponte HTTP elimina totalmente o gerenciamento de conexões.

Perguntas frequentes