Function Node do Node-RED: JavaScript para Mensagens MQTT

O function node do Node-RED permite executar seu próprio JavaScript em cada mensagem que passa pelo editor, o que faz dele a ferramenta mais flexível que você tem para remodelar dados MQTT. Neste guia você vai usá-lo para converter unidades, descartar leituras ruins, rotear mensagens para várias saídas, manter estado entre execuções e tratar erros de forma limpa, tudo contra um broker ao vivo que você acessa em segundos.

O que o function node do Node-RED realmente faz

Toda mensagem no Node-RED é um objeto JavaScript simples chamado msg. O function node recebe esse objeto, executa o código que você escreveu e repassa o que você retornar. Retorne o mesmo objeto e a mensagem segue em frente. Retorne null e a mensagem para. Retorne um array e cada elemento vai para uma porta de saída diferente. Esse é o contrato inteiro, e quase todo o resto é uma variação dele.

Dentro da função você tem acesso a um pequeno conjunto de auxiliares:

O editor também tem três abas: On Start roda uma vez quando o node é implantado, On Message roda para cada mensagem e On Stop roda quando o fluxo é parado ou reimplantado. On Start é o lugar certo para inicializar contadores, e On Stop é onde você limpa os timers que criou.

Se você ainda não montou um fluxo MQTT, comece pelo nosso tutorial de MQTT no Node-RED, que cobre em detalhes o node de configuração do broker e os nodes mqtt in e mqtt out.

Configure um feed de teste no servidor MQTT público

Você precisa de um fluxo de mensagens para trabalhar. O servidor MQTT público (gratuito) em broker.connectmq.com aceita conexões anônimas, então você pode publicar dados de teste a partir de um terminal sem nenhum cadastro. Escolha um prefixo exclusivo para que seus topics não colidam com os de mais ninguém:

# Publica uma leitura a cada dois segundos (Ctrl+C para parar)
while true; do
  T=$(awk -v r=$RANDOM 'BEGIN{printf "%.1f", 18 + (r % 150) / 10}')
  mosquitto_pub -h broker.connectmq.com -p 1883 -i demo-pub-7f3a \
    -t "connectmq/demo/7f3a/line1/sensor1/telemetry" \
    -m "{\"temp\": $T, \"unit\": \"C\", \"ts\": $(date +%s)}"
  sleep 2
done

No Node-RED, adicione um node mqtt in, aponte o servidor dele para broker.connectmq.com na porta 1883, dê ao node do broker um client ID exclusivo e não vazio, assine connectmq/demo/7f3a/# e defina Output como "a parsed JSON object". A partir daí, toda mensagem chega com msg.payload já convertido.

Lembre-se de que o servidor MQTT público pode ser lido por qualquer pessoa. Use-o apenas para testes e protótipos, nunca para dados reais de uma planta em Joinville, de uma fazenda ou de clientes.

Cinco exemplos de function node do Node-RED para MQTT

1. Transformar e enriquecer um payload

Os dispositivos raramente publicam exatamente o que o seu dashboard ou banco de dados precisa. Esta função converte Celsius para Fahrenheit, adiciona um rótulo de site e extrai o nome do dispositivo do topic:

// On Message
const parts = msg.topic.split("/");      // connectmq/demo/7f3a/line1/sensor1/telemetry
const line = parts[3];
const device = parts[4];

msg.payload = {
    device: device,
    line: line,
    tempC: msg.payload.temp,
    tempF: Math.round((msg.payload.temp * 9 / 5 + 32) * 10) / 10,
    receivedAt: new Date().toISOString()
};
return msg;

Dividir o topic é mais barato e mais confiável do que pedir a cada dispositivo que repita sua identidade dentro do payload.

2. Filtrar leituras ruins

Sensores falham. Um function node que retorna null descarta a mensagem silenciosamente, então você pode impor limites de sanidade antes que os dados cheguem ao armazenamento:

const t = msg.payload.temp;
if (typeof t !== "number" || Number.isNaN(t)) {
    node.warn("Non-numeric temp on " + msg.topic);
    return null;
}
if (t < -40 || t > 125) {
    return null;   // fora da faixa do sensor, descarta
}
return msg;

O node.warn escreve na barra lateral de debug e no log do Node-RED, o que ajuda a ver com que frequência a filtragem acontece sem adicionar nodes de debug extras.

3. Rotear para várias saídas

Defina Outputs como 2 nas configurações do node e retorne um array. A posição 0 vai para a primeira porta, a posição 1 para a segunda. Use null para uma porta que não deve disparar:

const limit = 30;
if (msg.payload.temp > limit) {
    const alert = {
        topic: msg.topic.replace("/telemetry", "/alert"),
        payload: { level: "high", temp: msg.payload.temp }
    };
    return [msg, alert];   // dados normais + alerta
}
return [msg, null];        // apenas dados normais

A segunda saída pode alimentar um node mqtt out com o campo de topic vazio, de modo que ele publique no que msg.topic indicar. Para roteamento simples por limiar, o node Switch costuma ser mais claro; recorra a uma função quando a condição envolver cálculos ou vários campos.

4. Manter estado entre mensagens

O contexto do node sobrevive entre mensagens enquanto o fluxo estiver rodando. Este exemplo publica apenas quando o valor muda mais de meio grau, uma banda morta simples que reduz muito o tráfego em sinais que variam devagar:

const key = "last_" + msg.topic;
const last = context.get(key);
const current = msg.payload.temp;

if (last !== undefined && Math.abs(current - last) < 0.5) {
    return null;
}
context.set(key, current);
return msg;

Se precisar do valor em outra aba ou depois de um restart, use flow, global ou um armazenamento persistente. Nosso guia sobre contexto no Node-RED explica os escopos e como habilitar o armazenamento em arquivo.

5. Enviar de forma assíncrona

Quando o seu código espera um timer ou uma promise, retornar da função é cedo demais. Use node.send() e chame node.done() ao terminar, para que os nodes Complete e Catch funcionem corretamente:

// Agrupa leituras e envia um resumo a cada 10 mensagens
let buf = context.get("buf") || [];
buf.push(msg.payload.temp);

if (buf.length >= 10) {
    const avg = buf.reduce((a, b) => a + b, 0) / buf.length;
    context.set("buf", []);
    setTimeout(() => {
        node.send({ topic: "connectmq/demo/7f3a/line1/sensor1/avg", payload: { avg: Number(avg.toFixed(2)) } });
        node.done();
    }, 0);
} else {
    context.set("buf", buf);
    node.done();
}
return null;

Um fluxo completo para importar

Copie este JSON, pressione Ctrl+I no editor, cole e faça o deploy. Depois rode o loop de publicação mostrado antes. Troque o prefixo 7f3a pela sua própria sequência aleatória nos dois lugares.

[
  {"id":"b1","type":"mqtt-broker","name":"ConnectMQ public","broker":"broker.connectmq.com","port":"1883","clientid":"nr-demo-7f3a","autoConnect":true,"usetls":false,"protocolVersion":"4","keepalive":"60","cleansession":true},
  {"id":"in1","type":"mqtt in","z":"t1","name":"telemetry","topic":"connectmq/demo/7f3a/+/+/telemetry","qos":"0","datatype":"json","broker":"b1","x":130,"y":100,"wires":[["fn1"]]},
  {"id":"fn1","type":"function","z":"t1","name":"enrich + alert","func":"const p = msg.topic.split('/');\nmsg.payload.device = p[4];\nmsg.payload.tempF = Math.round((msg.payload.temp * 9 / 5 + 32) * 10) / 10;\nif (msg.payload.temp > 30) {\n  return [msg, {topic: msg.topic.replace('/telemetry','/alert'), payload: {level:'high', temp: msg.payload.temp}}];\n}\nreturn [msg, null];","outputs":2,"x":330,"y":100,"wires":[["dbg1"],["out1"]]},
  {"id":"dbg1","type":"debug","z":"t1","name":"data","active":true,"complete":"payload","x":530,"y":80,"wires":[]},
  {"id":"out1","type":"mqtt out","z":"t1","name":"alerts","topic":"","qos":"1","retain":"false","broker":"b1","x":530,"y":130,"wires":[]},
  {"id":"t1","type":"tab","label":"Function demo"}
]

Tratamento de erros e depuração

Uma exceção lançada dentro de um function node registra um erro, mas, por padrão, a mensagem simplesmente desaparece. Dois hábitos tornam as falhas visíveis.

Primeiro, use node.error(err, msg) passando a mensagem como segundo argumento. Essa forma aciona qualquer node Catch na mesma aba, para que você possa encaminhar as falhas a um node de debug dedicado, a um arquivo de log ou a um topic de alerta.

try {
    msg.payload = JSON.parse(msg.payload.raw);
    return msg;
} catch (err) {
    node.error("Bad raw payload: " + err.message, msg);
    return null;
}

Segundo, use node.status() para mostrar diretamente no canvas o que o node está fazendo, por exemplo o último valor recebido ou a contagem de mensagens descartadas:

node.status({ fill: "green", shape: "dot", text: "last " + msg.payload.temp + " C" });

Checklist de erros comuns

1. Alterar uma mensagem que você também encaminha para outro lugar. Se uma saída vai para dois fios, o Node-RED clona a mensagem para você, mas objetos guardados no contexto são referências. Use RED.util.cloneMessage(msg) antes de armazenar.
2. Esquecer o return. Uma função sem instrução de retorno não envia nada, o que parece uma assinatura quebrada.
3. Usar uma string como objeto. Se o node mqtt in entrega uma string, msg.payload.temp é undefined. Configure a saída como JSON convertido ou adicione um node JSON; veja o guia do JSON node do Node-RED.
4. Deixar timers rodando. Limpe os intervals no On Stop, senão cada redeploy adiciona mais um.
5. Fazer trabalho pesado por mensagem. Um function node roda no mesmo event loop que o resto do Node-RED. Loops longos em topics de alta frequência deixam todos os fluxos lentos. Limite a taxa antes com o delay node do Node-RED.
6. Colocar segredos no código. Use env.get("API_KEY") em vez de colar tokens no corpo da função, porque arquivos de fluxo costumam ser compartilhados ou versionados no git.

Quando não usar um function node

O function node é poderoso, mas fluxos montados com vinte function nodes são difíceis de ler para os colegas. Prefira o node Change para definir e mover propriedades, o node Switch para rotear com condições simples e o node JSON para conversão. Reserve os function nodes para lógica que realmente precisa de código e dê a cada um nome claro. Se a mesma função aparecer em várias abas, empacote-a como subflow para corrigir bugs em um só lugar.

Você também pode carregar módulos npm em um function node pela aba Setup quando functionExternalModules estiver habilitado no settings.js. Isso é útil para coisas como bibliotecas de formatação de datas, mas cada módulo adicionado é mais uma dependência para atualizar e auditar. A documentação oficial do function node lista tudo o que está disponível na sandbox.

Rode seus fluxos Node-RED no ConnectMQ

Quando a sua lógica de função funcionar no servidor MQTT público, mova-a para um broker privado editando um único node de configuração. O ConnectMQ é um broker MQTT gerenciado com TLS na porta 8883, MQTT 3.1.1 e 5.0 e um portal com cliente no navegador e explorador de topics ao vivo, para você acompanhar os alertas que sua função publica. Crie um broker Serverless gratuito em app.connectmq.com sem cartão de crédito, copie o username, a senha e o client ID gerados para o node do broker no Node-RED, marque Use TLS e faça o redeploy. Se estiver comparando opções de hospedagem, nossa visão geral de um broker MQTT na nuvem explica o que observar.

Perguntas frequentes

Um function node do Node-RED pode usar async/await?

Sim. O corpo da função roda dentro de um wrapper assíncrono, então você pode usar await em promises. Envie os resultados com node.send() e chame node.done(), em vez de depender do valor de retorno após o await.

Como envio várias mensagens a partir de um function node?

Retorne um array de mensagens para saídas diferentes, ou um array dentro de outro array, como return [[m1, m2, m3]], para enviar várias mensagens em sequência pela primeira saída.

Por que meu function node não envia nada?

Normalmente a função retorna null ou undefined, ou uma exceção foi lançada. Verifique os erros na barra lateral de debug e adicione node.warn(msg) no início para confirmar que as mensagens estão chegando.

O function node é lento para dados MQTT de alta frequência?

Transformações simples são rápidas, mas toda função roda no único event loop do Node.js. Para taxas de mensagens muito altas, mantenha o código leve e filtre ou limite a taxa logo no início do fluxo.