Tutorial de Cliente MQTT em Ruby: Publish, Subscribe e TLS com a Gem mqtt

Um cliente MQTT em Ruby permite que a sua aplicação Rails ou Ruby puro converse com dispositivos em tempo real com muito pouco código. Neste tutorial você vai usar a gem mqtt para publicar e assinar, ativar a verificação correta de certificados TLS, construir um assinante que sobrevive a quedas de rede e integrá-lo a um app Rails como worker em segundo plano.

A gem mqtt em resumo

A gem mqtt (muitas vezes chamada de ruby-mqtt) é uma implementação em Ruby puro do protocolo cliente MQTT 3.1.1. Ela é pequena, não tem extensões nativas e sua API se lê como Ruby comum. Algumas características definem como você a usa:

Se você precisa de MQTT 5 no Ruby ou quer comparar opções, o guia de bibliotecas cliente MQTT apresenta o panorama. Para a maioria das integrações com Rails, o 3.1.1 é tudo de que você precisa, e o ConnectMQ aceita as duas versões.

Instale a gem:

gem install mqtt
# ou em um app Rails
bundle add mqtt

Primeiro cliente MQTT em Ruby: publish e subscribe

Comece com o servidor MQTT público (gratuito) em broker.connectmq.com, feito para testes e protótipos. Ele aceita conexões anônimas com qualquer client ID não vazio. Tudo na área pública pode ser lido por qualquer pessoa, então use um prefixo de topic aleatório e apenas dados de teste.

hello.rb:

require "mqtt"
require "securerandom"

PREFIX = "connectmq/demo/ruby-#{SecureRandom.hex(4)}"

subscriber = Thread.new do
  MQTT::Client.connect(host: "broker.connectmq.com", port: 1883,
                       client_id: "ruby-sub-#{SecureRandom.hex(4)}") do |c|
    c.get("#{PREFIX}/#") do |topic, message|
      puts "received #{topic}: #{message}"
      break if message == "bye"
    end
  end
end

sleep 1 # dá tempo para o assinante assinar

MQTT::Client.connect(host: "broker.connectmq.com", port: 1883,
                     client_id: "ruby-pub-#{SecureRandom.hex(4)}") do |c|
  c.publish("#{PREFIX}/greeting", "hello from Ruby")
  c.publish("#{PREFIX}/greeting", "bye")
end

subscriber.join

Execute com ruby hello.rb. A forma com bloco de MQTT::Client.connect desconecta automaticamente quando o bloco termina, o jeito mais limpo de fazer publicações de curta duração. A porta 1883 não é criptografada, o que serve para este primeiro teste e para nada além disso.

Ativando TLS com verificação de certificado

Para qualquer coisa além de um teste rápido, conecte na porta 8883 com TLS. Definir apenas ssl: true criptografa a conexão, mas a verificação do certificado do servidor depende do contexto SSL. Seja explícito e configure a verificação você mesmo:

tls_client.rb:

require "mqtt"
require "openssl"
require "securerandom"

def build_client(client_id: nil)
  client = MQTT::Client.new
  client.host = ENV.fetch("MQTT_HOST", "broker.connectmq.com")
  client.port = 8883
  client.ssl = true
  client.ssl_context.set_params(verify_mode: OpenSSL::SSL::VERIFY_PEER)
  client.username = ENV["MQTT_USERNAME"]          # nil = anônimo (broker público)
  client.password = ENV["MQTT_PASSWORD"]
  client.client_id = client_id || ENV["MQTT_CLIENT_ID"] || "ruby-#{SecureRandom.hex(4)}"
  client.keep_alive = 30
  client.clean_session = true
  client
end

if $PROGRAM_NAME == __FILE__
  client = build_client
  client.connect
  client.publish("connectmq/demo/ruby-tls-test/status", "online", false, 1)
  puts "published over TLS as #{client.client_id}"
  client.disconnect
end

set_params carrega o repositório de certificados padrão do Ruby e ativa a verificação do hostname, de modo que um certificado de man-in-the-middle emitido para outro host é rejeitado. O ConnectMQ serve um certificado Let's Encrypt publicamente confiável na 8883, então nenhum arquivo de CA personalizado é necessário. Se o seu Ruby foi compilado contra um OpenSSL sem bundle de CAs (comum em imagens Docker enxutas), instale ca-certificates na imagem. Mais detalhes sobre certificados estão em configuração de TLS no MQTT.

Os argumentos posicionais de publish são topic, payload, retain, qos. QoS 1 faz o broker confirmar a mensagem, que é o que você quer para comandos.

Um assinante que reconecta

Como a gem não reconecta, um assinante de longa duração precisa de um loop que capture erros de rede, espere com backoff e recomece. Ele também precisa refazer a assinatura a cada vez, porque uma clean session esquece as assinaturas.

mqtt_worker.rb:

require "json"
require_relative "tls_client"

class MqttWorker
  RETRYABLE = [MQTT::Exception, SocketError, IOError, SystemCallError,
               OpenSSL::SSL::SSLError, Timeout::Error].freeze

  def initialize(filter:, handler:)
    @filter = filter
    @handler = handler
    @running = true
  end

  def stop
    @running = false
    @client&.disconnect rescue nil
  end

  def run
    backoff = 1
    while @running
      begin
        @client = build_client
        @client.connect
        @client.subscribe(@filter => 1)
        warn "connected, subscribed to #{@filter}"
        backoff = 1

        @client.get do |topic, payload|
          handle(topic, payload)
          break unless @running
        end
      rescue *RETRYABLE => e
        break unless @running
        warn "mqtt error: #{e.class}: #{e.message}; retrying in #{backoff}s"
        sleep backoff
        backoff = [backoff * 2, 60].min
      ensure
        @client&.disconnect rescue nil
      end
    end
  end

  private

  def handle(topic, payload)
    data = JSON.parse(payload)
    @handler.call(topic, data)
  rescue JSON::ParserError
    warn "ignoring non-JSON payload on #{topic}"
  rescue StandardError => e
    warn "handler failed for #{topic}: #{e.class}: #{e.message}"
  end
end

if $PROGRAM_NAME == __FILE__
  prefix = ENV.fetch("MQTT_TOPIC_PREFIX", "connectmq/demo/ruby-worker-7c2e")
  worker = MqttWorker.new(
    filter: "#{prefix}/devices/+/telemetry",
    handler: ->(topic, data) { puts "#{topic.split('/')[-2]}: #{data.inspect}" }
  )
  %w[INT TERM].each { |sig| trap(sig) { Thread.new { worker.stop } } }
  worker.run
end

Dois detalhes passam despercebidos com facilidade. Primeiro, os erros do handler são resgatados dentro de handle, então uma mensagem ruim não derruba a conexão. Segundo, handlers de trap no Ruby não podem fazer muito trabalho com segurança, então a parada acontece em uma nova thread; stop fecha o socket, get lança uma exceção ou retorna, e o loop termina.

Teste com um dispositivo simulado:

ruby mqtt_worker.rb

# em outro terminal
mosquitto_pub -h broker.connectmq.com -p 1883 -i ruby-sim-boiler \
  -t "connectmq/demo/ruby-worker-7c2e/devices/boiler1/telemetry" \
  -m '{"temp": 71.5, "burner": true}'

Usando o cliente MQTT em Ruby com Rails

Apps Rails precisam de duas coisas: publicar de dentro do app e um worker que consome as mensagens dos dispositivos para o banco de dados.

Publicando a partir do Rails

Mantenha a publicação em um pequeno service object e chame-o a partir de um job em segundo plano, não diretamente em um controller, para que uma rede lenta nunca bloqueie uma requisição web:

# app/services/device_commands.rb
require Rails.root.join("lib/tls_client")

class DeviceCommands
  def self.send!(device_id, action)
    raise ArgumentError, "bad device" unless device_id.to_s.match?(/\A[a-z0-9_-]+\z/i)

    topic = "#{ENV.fetch('MQTT_TOPIC_PREFIX')}/devices/#{device_id}/set"
    client = build_client(client_id: "#{ENV.fetch('MQTT_CLIENT_ID_PREFIX', 'rails')}-pub-#{SecureRandom.hex(3)}")
    client.connect
    client.publish(topic, { action: action, ts: Time.now.to_i }.to_json, false, 1)
  ensure
    client&.disconnect
  end
end

# app/jobs/device_command_job.rb
class DeviceCommandJob < ApplicationJob
  queue_as :default
  def perform(device_id, action) = DeviceCommands.send!(device_id, action)
end

A verificação do ID do dispositivo impede que usuários injetem +, # ou / no topic e endereçem dispositivos que não são deles. Em um broker privado onde os client IDs ficam vinculados às credenciais emitidas, use exatamente o client ID que você recebeu em vez de um sufixo aleatório, e direcione toda a publicação por um único worker se você tiver apenas um conjunto de credenciais.

O worker consumidor

Coloque tls_client.rb e mqtt_worker.rb em lib/ e adicione um script de execução, bin/mqtt_worker, que inicializa o Rails para que o handler possa usar os models:

#!/usr/bin/env ruby
require_relative "../config/environment"
require Rails.root.join("lib/mqtt_worker")

worker = MqttWorker.new(
  filter: "#{ENV.fetch('MQTT_TOPIC_PREFIX')}/devices/+/telemetry",
  handler: lambda do |topic, data|
    device_id = topic.split("/")[-2]
    Reading.create!(device_id: device_id, payload: data, received_at: Time.current)
  end
)
%w[INT TERM].each { |sig| trap(sig) { Thread.new { worker.stop } } }
worker.run

Rode-o como um tipo de processo próprio, por exemplo em um Procfile:

web: bundle exec puma -C config/puma.rb
mqtt: bundle exec ruby bin/mqtt_worker

Rode exatamente uma instância do worker por client ID. Dois processos com o mesmo ID fazem o broker desconectar o mais antigo, e eles ficam derrubando um ao outro. Se você precisa de vários consumidores dividindo a carga, as shared subscriptions do MQTT 5 são a resposta usual, mas exigem um cliente MQTT 5; o guia de shared subscriptions explica o padrão.

Publicando com eficiência em taxas mais altas

Abrir uma conexão TLS a cada publicação é simples, mas cada uma custa um handshake e uma nova sessão no broker. Para alguns comandos por minuto, isso é irrelevante. Para centenas de mensagens por minuto, mantenha uma conexão aberta em um processo de longa duração e alimente-a a partir de uma fila. Uma thread publicadora dedicada que retira itens de uma Queue do Ruby, publica e reconecta em caso de erro com o mesmo backoff do worker dá a você uma conexão estável e um único client ID.

Canalize todas as publicações por essa única thread em vez de compartilhar um objeto cliente entre muitas threads. A gem não documenta thread safety para publicações concorrentes, e serializar as escritas por uma fila evita pacotes intercalados no socket. Isso também cria um lugar natural para aplicar backpressure: se a fila crescer além de um limite, rejeite trabalho novo ou descarte telemetria de baixa prioridade, em vez de deixar a memória crescer enquanto o broker está inacessível.

Checklist de produção

1. Crie um broker privado em um broker MQTT gerenciado e copie o usuário, a senha e o client ID gerados.
2. Guarde MQTT_USERNAME, MQTT_PASSWORD, MQTT_CLIENT_ID e MQTT_TOPIC_PREFIX nas credentials do Rails ou no cofre de segredos da sua plataforma.
3. Conecte na porta 8883 com VERIFY_PEER; nunca coloque VERIFY_NONE em produção.
4. Mantenha o worker como um processo separado e supervisionado, com reinício automático.
5. Torne o handler idempotente: o QoS 1 pode entregar uma mensagem mais de uma vez.
6. Valide qualquer entrada do usuário que acabe em um nome de topic.

Solução de problemas

Para outras linguagens de back-end com os mesmos padrões, veja o tutorial de cliente MQTT em PHP e o guia sobre como testar clientes MQTT.

Comece a usar o ConnectMQ

Conecte o seu app Ruby a um broker privado e protegido por TLS em poucos minutos. Cadastre-se em app.connectmq.com para obter um broker Serverless gratuito, sem cartão de crédito, defina as credenciais geradas como variáveis de ambiente e acompanhe os topics do seu worker no explorador de topics ao vivo do portal. O ConnectMQ opera o broker, então a sua equipe só mantém código Ruby.

Perguntas frequentes

A gem mqtt do Ruby suporta MQTT 5?

Não, ela implementa o MQTT 3.1.1. Isso é totalmente compatível com o ConnectMQ, que suporta tanto o 3.1.1 quanto o 5.0.

O cliente MQTT em Ruby reconecta automaticamente?

Não. Envolva connect, subscribe e get em um loop que resgata erros de rede, espera com backoff e refaz a assinatura, como no worker acima.

Posso assinar MQTT dentro de uma requisição web do Rails?

Não deveria. Assinaturas precisam de um processo de longa duração, então rode um worker separado e reserve as requisições web para publicações rápidas por meio de jobs em segundo plano.

Como verificar o certificado do broker no Ruby?

Defina client.ssl = true e chame client.ssl_context.set_params(verify_mode: OpenSSL::SSL::VERIFY_PEER), que usa o repositório de CAs do sistema e verifica o hostname.