Tutorial de cliente MQTT en Ruby: publicar, suscribirse y TLS con la gema mqtt

Un cliente MQTT en Ruby permite que tu aplicación Rails o Ruby pura hable con dispositivos en tiempo real con muy poco código. En este tutorial usarás la gema mqtt para publicar y suscribirte, activarás una verificación adecuada de certificados TLS, construirás un suscriptor que sobrevive a los cortes de red y lo integrarás en una app Rails como worker en segundo plano.

La gema mqtt de un vistazo

La gema mqtt (a menudo llamada ruby-mqtt) es una implementación en Ruby puro del protocolo cliente MQTT 3.1.1. Es pequeña, no tiene extensiones nativas y su API se lee como Ruby normal. Algunas características condicionan cómo la usas:

Si necesitas MQTT 5 desde Ruby o quieres comparar opciones, la guía de bibliotecas cliente MQTT repasa el panorama. Para la mayoría de las integraciones con Rails, 3.1.1 es todo lo que necesitas, y ConnectMQ acepta ambas versiones.

Instala la gema:

gem install mqtt
# o en una app Rails
bundle add mqtt

Primer cliente MQTT en Ruby: publicar y suscribirse

Empieza con el servidor MQTT público gratuito en broker.connectmq.com. Acepta conexiones anónimas con cualquier client ID no vacío. Todo lo que hay en el área pública es legible por cualquiera, así que usa un prefijo de topic aleatorio y solo datos de prueba: es para pruebas y prototipos, no para datos de producción.

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 # dar tiempo al suscriptor para suscribirse

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

Ejecútalo con ruby hello.rb. La forma con bloque de MQTT::Client.connect desconecta automáticamente al terminar el bloque, que es la manera más limpia de hacer publicaciones breves. El puerto 1883 no está cifrado, lo cual vale para esta primera prueba y para nada más.

Activar TLS con verificación de certificados

Para cualquier cosa que vaya más allá de una prueba rápida, conéctate al puerto 8883 con TLS. Definir solo ssl: true cifra la conexión, pero que se verifique o no el certificado del servidor depende del contexto SSL. Sé explícito y configura tú mismo la verificación:

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 carga el almacén de certificados por defecto de Ruby y activa la verificación del nombre de host, de modo que se rechaza un certificado de ataque man-in-the-middle emitido para otro host. ConnectMQ sirve un certificado de Let's Encrypt de confianza pública en el 8883, así que no hace falta ningún archivo de CA personalizado. Si tu Ruby se compiló contra un OpenSSL sin paquete de CA (habitual en imágenes Docker slim), instala ca-certificates en la imagen. Tienes más detalles sobre certificados en configuración TLS de MQTT.

Los argumentos posicionales de publish son topic, payload, retain, qos. QoS 1 hace que el broker confirme el mensaje, que es lo que quieres para los comandos.

Un suscriptor que reconecta

Como la gema no reconecta, un suscriptor de larga duración necesita un bucle que capture los errores de red, espere con backoff y vuelva a empezar. También necesita volver a suscribirse cada vez, porque una sesión limpia olvida las suscripciones.

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

Hay dos detalles fáciles de pasar por alto. Primero, los errores del handler se capturan dentro de handle, de modo que un mensaje defectuoso no puede tumbar la conexión. Segundo, los handlers de trap en Ruby no pueden hacer mucho trabajo de forma segura, así que la parada ocurre en un hilo nuevo; stop cierra el socket, get lanza una excepción o retorna, y el bucle termina.

Pruébalo con un dispositivo simulado:

ruby mqtt_worker.rb

# en otra 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}'

Usar el cliente MQTT de Ruby con Rails

Las apps Rails necesitan dos cosas: publicar desde dentro de la app y un worker que consuma los mensajes de los dispositivos y los lleve a la base de datos.

Publicar desde Rails

Mantén la publicación en un pequeño service object y llámalo desde un job en segundo plano, no directamente en un controlador, para que una red lenta nunca bloquee una petición 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

La comprobación del ID de dispositivo impide que los usuarios inyecten +, # o / en el topic y se dirijan a dispositivos que no son suyos. En un broker privado donde los client ID están vinculados a las credenciales emitidas, usa exactamente el client ID que se te asignó en lugar de un sufijo aleatorio, y canaliza todas las publicaciones a través de un único worker si solo tienes un conjunto de credenciales.

El worker consumidor

Pon tls_client.rb y mqtt_worker.rb en lib/ y añade un script de arranque, bin/mqtt_worker, que inicie Rails para que el handler pueda usar los modelos:

#!/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

Ejecútalo como su propio tipo de proceso, por ejemplo en un Procfile:

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

Ejecuta exactamente una instancia del worker por client ID. Dos procesos con el mismo ID hacen que el broker desconecte el más antiguo, y se expulsan mutuamente sin parar. Si necesitas varios consumidores que se repartan la carga, la respuesta habitual son las suscripciones compartidas de MQTT 5, pero requieren un cliente MQTT 5; la guía de suscripciones compartidas explica el patrón.

Publicar de forma eficiente a mayor ritmo

Abrir una conexión TLS por cada publicación es sencillo, pero cada una cuesta un handshake y una sesión nueva en el broker. Para unos pocos comandos por minuto eso es irrelevante. Para cientos de mensajes por minuto, mantén una conexión abierta en un proceso de larga duración y aliméntala desde una cola. Un hilo publicador dedicado que saca elementos de una Queue de Ruby, publica y reconecta ante errores con el mismo backoff que el worker te da una conexión estable y un único client ID.

Canaliza todas las publicaciones a través de ese único hilo en lugar de compartir un objeto cliente entre muchos hilos. La gema no documenta la seguridad entre hilos para publicaciones concurrentes, y serializar las escrituras mediante una cola evita paquetes entrelazados en el socket. Además, te da un lugar natural para aplicar contrapresión: si la cola supera un límite, rechaza el trabajo nuevo o descarta la telemetría de baja prioridad en lugar de dejar crecer la memoria mientras el broker no está accesible.

Lista de comprobación para producción

1. Crea un broker privado en un broker MQTT gestionado y copia el usuario, la contraseña y el client ID generados.
2. Guarda MQTT_USERNAME, MQTT_PASSWORD, MQTT_CLIENT_ID y MQTT_TOPIC_PREFIX en las credenciales de Rails o en el almacén de secretos de tu plataforma.
3. Conéctate al puerto 8883 con VERIFY_PEER; nunca publiques con VERIFY_NONE.
4. Mantén el worker como un proceso supervisado independiente con reinicios automáticos.
5. Haz que el handler sea idempotente: QoS 1 puede entregar un mensaje más de una vez.
6. Valida cualquier entrada de usuario que acabe en el nombre de un topic.

Solución de problemas

Para otros lenguajes de backend con los mismos patrones, consulta el tutorial de cliente MQTT en PHP y la guía de pruebas de clientes MQTT.

Empieza con ConnectMQ

Conecta tu app Ruby a un broker privado protegido con TLS en pocos minutos. Regístrate en app.connectmq.com para obtener un broker Serverless gratuito sin tarjeta de crédito, define las credenciales generadas como variables de entorno y observa los topics de tu worker en el explorador de topics en vivo del portal. ConnectMQ opera el broker, así que tu equipo solo mantiene código Ruby.

Preguntas frecuentes

¿La gema mqtt de Ruby es compatible con MQTT 5?

No, implementa MQTT 3.1.1. Es totalmente compatible con ConnectMQ, que admite tanto 3.1.1 como 5.0.

¿El cliente MQTT de Ruby reconecta automáticamente?

No. Envuelve connect, subscribe y get en un bucle que capture los errores de red, espere con backoff y vuelva a suscribirse, como en el worker anterior.

¿Puedo suscribirme a MQTT dentro de una petición web de Rails?

No deberías. Las suscripciones necesitan un proceso de larga duración, así que ejecuta un worker independiente y limita las peticiones web a publicaciones rápidas mediante jobs en segundo plano.

¿Cómo verifico el certificado del broker en Ruby?

Define client.ssl = true y llama a client.ssl_context.set_params(verify_mode: OpenSSL::SSL::VERIFY_PEER), que usa el almacén de CA del sistema y comprueba el nombre de host.