Ruby MQTT Client Tutorial: Publish, Subscribe and TLS with the mqtt Gem

A Ruby MQTT client lets your Rails or plain Ruby application talk to devices in real time with very little code. In this tutorial you will use the mqtt gem to publish and subscribe, turn on proper TLS certificate verification, build a subscriber that survives network drops, and wire it into a Rails app as a background worker.

The mqtt Gem at a Glance

The mqtt gem (often called ruby-mqtt) is a pure Ruby implementation of the MQTT 3.1.1 client protocol. It is small, has no native extensions, and its API reads like plain Ruby. A few properties shape how you use it:

If you need MQTT 5 from Ruby or want to compare options, the MQTT client libraries guide covers the landscape. For most Rails integrations, 3.1.1 is all you need, and ConnectMQ accepts both versions.

Install the gem:

gem install mqtt
# or in a Rails app
bundle add mqtt

First Ruby MQTT Client: Publish and Subscribe

Start with the free public MQTT broker at broker.connectmq.com. It accepts anonymous connections with any non-empty client ID. Everything in the public area is readable by anyone, so use a random topic prefix and test data only.

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 # give the subscriber time to subscribe

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

Run it with ruby hello.rb. The block form of MQTT::Client.connect disconnects automatically when the block ends, which is the cleanest way to do short-lived publishing. Port 1883 is unencrypted, which is fine for this first test and nothing else.

Turning On TLS With Certificate Verification

For anything beyond a quick test, connect on port 8883 with TLS. Setting ssl: true alone encrypts the connection, but whether the server certificate is verified depends on the SSL context. Be explicit and configure verification yourself:

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 = anonymous (public broker)
  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 loads Ruby's default certificate store and enables hostname verification, so a man-in-the-middle certificate for another host is rejected. ConnectMQ serves a publicly trusted Let's Encrypt certificate on 8883, so no custom CA file is needed. If your Ruby was compiled against an OpenSSL without a CA bundle (common in slim Docker images), install ca-certificates in the image. More detail on certificates is in MQTT TLS configuration.

The positional arguments to publish are topic, payload, retain, qos. QoS 1 makes the broker acknowledge the message, which is what you want for commands.

A Reconnecting Subscriber

Because the gem does not reconnect, a long-running subscriber needs a loop that catches network errors, waits with backoff and starts again. It also needs to resubscribe each time, because a clean session forgets subscriptions.

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

Two details are easy to miss. First, handler errors are rescued inside handle, so one bad message cannot tear down the connection. Second, trap handlers in Ruby cannot safely do much work, so the stop happens in a new thread; stop closes the socket, get raises or returns, and the loop exits.

Test it with a simulated device:

ruby mqtt_worker.rb

# in another 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}'

Using the Ruby MQTT Client With Rails

Rails apps need two things: publishing from inside the app, and a worker that consumes device messages into the database.

Publishing From Rails

Keep publishing in a small service object and call it from a background job, not directly in a controller, so a slow network never blocks a web request:

# 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

The device ID check prevents users from injecting +, # or / into the topic and addressing devices that are not theirs. On a private broker where client IDs are bound to issued credentials, use the exact client ID you were given instead of a random suffix, and route all publishing through a single worker if you only have one credential set.

The Consumer Worker

Put tls_client.rb and mqtt_worker.rb in lib/, then add a runner script, bin/mqtt_worker, that boots Rails so the handler can use 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

Run it as its own process type, for example in a Procfile:

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

Run exactly one instance of the worker per client ID. Two processes with the same ID make the broker disconnect the older one, and they keep kicking each other off. If you need several consumers sharing the load, MQTT 5 shared subscriptions are the usual answer, but they need an MQTT 5 client; the shared subscriptions guide explains the pattern.

Publishing Efficiently at Higher Rates

Opening a TLS connection for every publish is simple, but each one costs a handshake and a new session on the broker. For a few commands per minute that is irrelevant. For hundreds of messages per minute, hold one connection open in a long-lived process and feed it from a queue. A dedicated publisher thread that pops from a Ruby Queue, publishes, and reconnects on errors with the same backoff as the worker gives you one stable connection and one client ID.

Funnel all publishes through that single thread rather than sharing one client object across many threads. The gem does not document thread safety for concurrent publishes, and serialising writes through a queue avoids interleaved packets on the socket. It also gives you a natural place to apply backpressure: if the queue grows beyond a limit, reject new work or drop low-priority telemetry instead of letting memory grow while the broker is unreachable.

Production Checklist

1. Create a private broker on a managed MQTT broker and copy the generated username, password and client ID.
2. Store MQTT_USERNAME, MQTT_PASSWORD, MQTT_CLIENT_ID and MQTT_TOPIC_PREFIX in Rails credentials or your platform's secret store.
3. Connect on port 8883 with VERIFY_PEER; never ship VERIFY_NONE.
4. Keep the worker as a separate supervised process with automatic restarts.
5. Make the handler idempotent: QoS 1 can deliver a message more than once.
6. Validate any user input that ends up in a topic name.

Troubleshooting

For other back-end languages with the same patterns, see the PHP MQTT client tutorial and the guide to testing MQTT clients.

Get Started with ConnectMQ

Connect your Ruby app to a private, TLS-secured broker in a few minutes. Sign up at app.connectmq.com for a free Serverless broker with no credit card, set the generated credentials as environment variables, and watch your worker's topics in the portal's live topic explorer. ConnectMQ operates the broker, so your team only maintains Ruby code.

Frequently Asked Questions

Does the Ruby mqtt gem support MQTT 5?

No, it implements MQTT 3.1.1. That is fully compatible with ConnectMQ, which supports both 3.1.1 and 5.0.

Does the Ruby MQTT client reconnect automatically?

No. Wrap connect, subscribe and get in a loop that rescues network errors, sleeps with backoff and resubscribes, as in the worker above.

Can I subscribe to MQTT inside a Rails web request?

You should not. Subscriptions need a long-lived process, so run a separate worker and keep web requests to quick publishes via background jobs.

How do I verify the broker certificate in Ruby?

Set client.ssl = true and call client.ssl_context.set_params(verify_mode: OpenSSL::SSL::VERIFY_PEER), which uses the system CA store and checks the hostname.