Tutoriel client MQTT Ruby : publish, subscribe et TLS avec la gem mqtt

Un client MQTT Ruby permet à votre application Rails ou Ruby pure de communiquer en temps réel avec des appareils, avec très peu de code. Dans ce tutoriel, vous allez utiliser la gem mqtt pour publier et vous abonner, activer une vérification correcte des certificats TLS, construire un subscriber qui survit aux coupures réseau et l'intégrer à une application Rails sous forme de worker d'arrière-plan.

La gem mqtt en un coup d'œil

La gem mqtt (souvent appelée ruby-mqtt) est une implémentation Ruby pure du protocole client MQTT 3.1.1. Elle est légère, sans extension native, et son API se lit comme du Ruby ordinaire. Quelques caractéristiques déterminent la façon de l'utiliser :

Si vous avez besoin de MQTT 5 en Ruby ou voulez comparer les options, le guide des bibliothèques clientes MQTT fait le tour du sujet. Pour la plupart des intégrations Rails, la version 3.1.1 suffit, et ConnectMQ accepte les deux versions.

Installez la gem :

gem install mqtt
# ou dans une application Rails
bundle add mqtt

Premier client MQTT Ruby : publish et subscribe

Commencez avec le serveur MQTT public gratuit à l'adresse broker.connectmq.com. Il accepte les connexions anonymes avec n'importe quel client ID non vide. Tout ce qui se trouve dans la zone publique est lisible par n'importe qui : utilisez donc un préfixe de topic aléatoire et uniquement des données de test.

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 # laisser au subscriber le temps de s'abonner

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

Exécutez-le avec ruby hello.rb. La forme bloc de MQTT::Client.connect se déconnecte automatiquement à la fin du bloc, ce qui est la manière la plus propre de publier ponctuellement. Le port 1883 n'est pas chiffré, ce qui convient pour ce premier test et rien d'autre.

Activer TLS avec vérification des certificats

Au-delà d'un test rapide, connectez-vous sur le port 8883 en TLS. Définir ssl: true seul chiffre la connexion, mais la vérification du certificat du serveur dépend du contexte SSL. Soyez explicite et configurez vous-même la vérification :

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 = anonyme (broker public)
  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 charge le magasin de certificats par défaut de Ruby et active la vérification du nom d'hôte : un certificat d'attaque de l'homme du milieu émis pour un autre hôte est donc rejeté. ConnectMQ sert un certificat Let's Encrypt reconnu publiquement sur 8883, aucun fichier de CA personnalisé n'est donc nécessaire. Si votre Ruby a été compilé avec un OpenSSL sans bundle de CA (fréquent dans les images Docker allégées), installez ca-certificates dans l'image. Plus de détails sur les certificats dans la configuration TLS de MQTT.

Les arguments positionnels de publish sont topic, payload, retain, qos. Le QoS 1 oblige le broker à acquitter le message, ce que vous voulez pour des commandes.

Un subscriber qui se reconnecte

Comme la gem ne se reconnecte pas, un subscriber de longue durée a besoin d'une boucle qui intercepte les erreurs réseau, attend avec un backoff et redémarre. Il doit aussi se réabonner à chaque fois, car une clean session oublie les abonnements.

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

Deux détails passent facilement inaperçus. D'abord, les erreurs du handler sont interceptées dans handle, de sorte qu'un mauvais message ne peut pas faire tomber la connexion. Ensuite, les handlers trap en Ruby ne peuvent pas effectuer beaucoup de travail en toute sécurité ; l'arrêt se fait donc dans un nouveau thread : stop ferme le socket, get lève une exception ou retourne, et la boucle se termine.

Testez-le avec un appareil simulé :

ruby mqtt_worker.rb

# dans un autre 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}'

Utiliser le client MQTT Ruby avec Rails

Les applications Rails ont besoin de deux choses : publier depuis l'application, et un worker qui consomme les messages des appareils pour les enregistrer en base de données.

Publier depuis Rails

Gardez la publication dans un petit objet de service et appelez-le depuis un job d'arrière-plan, pas directement dans un contrôleur, pour qu'un réseau lent ne bloque jamais une requête 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 vérification de l'ID d'appareil empêche les utilisateurs d'injecter +, # ou / dans le topic et d'adresser des appareils qui ne leur appartiennent pas. Sur un broker privé où les client IDs sont liés aux identifiants émis, utilisez le client ID exact qui vous a été attribué au lieu d'un suffixe aléatoire, et faites passer toutes les publications par un worker unique si vous ne disposez que d'un seul jeu d'identifiants.

Le worker consommateur

Placez tls_client.rb et mqtt_worker.rb dans lib/, puis ajoutez un script de lancement, bin/mqtt_worker, qui démarre Rails pour que le handler puisse utiliser les modèles :

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

Exécutez-le comme un type de processus à part entière, par exemple dans un Procfile :

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

Exécutez exactement une instance du worker par client ID. Deux processus avec le même ID poussent le broker à déconnecter le plus ancien, et ils s'éjectent mutuellement sans fin. Si vous avez besoin de plusieurs consommateurs se partageant la charge, les abonnements partagés de MQTT 5 sont la réponse habituelle, mais ils nécessitent un client MQTT 5 ; le guide des abonnements partagés explique ce modèle.

Publier efficacement à des débits plus élevés

Ouvrir une connexion TLS pour chaque publication est simple, mais chacune coûte une négociation et une nouvelle session sur le broker. Pour quelques commandes par minute, c'est sans importance. Pour des centaines de messages par minute, maintenez une connexion ouverte dans un processus de longue durée et alimentez-la depuis une file. Un thread de publication dédié qui dépile une Queue Ruby, publie et se reconnecte en cas d'erreur avec le même backoff que le worker vous donne une connexion stable et un client ID unique.

Faites passer toutes les publications par ce thread unique plutôt que de partager un objet client entre de nombreux threads. La gem ne documente pas la sûreté des publications concurrentes entre threads, et sérialiser les écritures via une file évite l'entrelacement des paquets sur le socket. Cela vous donne aussi un endroit naturel pour appliquer une contre-pression : si la file dépasse une limite, refusez le nouveau travail ou abandonnez la télémétrie de faible priorité au lieu de laisser la mémoire grossir pendant que le broker est injoignable.

Liste de contrôle pour la production

1. Créez un broker privé sur un broker MQTT managé et copiez le username, le mot de passe et le client ID générés.
2. Stockez MQTT_USERNAME, MQTT_PASSWORD, MQTT_CLIENT_ID et MQTT_TOPIC_PREFIX dans les credentials Rails ou le gestionnaire de secrets de votre plateforme.
3. Connectez-vous sur le port 8883 avec VERIFY_PEER ; ne livrez jamais VERIFY_NONE.
4. Gardez le worker dans un processus séparé et supervisé avec redémarrages automatiques.
5. Rendez le handler idempotent : le QoS 1 peut livrer un message plusieurs fois.
6. Validez toute entrée utilisateur qui se retrouve dans un nom de topic.

Dépannage

Pour d'autres langages back-end appliquant les mêmes modèles, consultez le tutoriel du client MQTT PHP et le guide pour tester les clients MQTT.

Démarrer avec ConnectMQ

Connectez votre application Ruby à un broker privé sécurisé par TLS en quelques minutes. Inscrivez-vous sur app.connectmq.com pour obtenir un broker Serverless gratuit sans carte bancaire, définissez les identifiants générés comme variables d'environnement et suivez les topics de votre worker dans l'explorateur de topics en direct du portail. ConnectMQ exploite le broker, votre équipe ne maintient donc que du code Ruby.

Questions fréquentes

La gem Ruby mqtt prend-elle en charge MQTT 5 ?

Non, elle implémente MQTT 3.1.1. C'est entièrement compatible avec ConnectMQ, qui prend en charge 3.1.1 comme 5.0.

Le client MQTT Ruby se reconnecte-t-il automatiquement ?

Non. Encapsulez connect, subscribe et get dans une boucle qui intercepte les erreurs réseau, attend avec un backoff et se réabonne, comme dans le worker ci-dessus.

Puis-je m'abonner à MQTT dans une requête web Rails ?

Vous ne devriez pas. Les abonnements nécessitent un processus de longue durée : exécutez un worker séparé et limitez les requêtes web à des publications rapides via des jobs d'arrière-plan.

Comment vérifier le certificat du broker en Ruby ?

Définissez client.ssl = true et appelez client.ssl_context.set_params(verify_mode: OpenSSL::SSL::VERIFY_PEER), qui utilise le magasin de CA du système et vérifie le nom d'hôte.