MQTT Ruby + Adafruit IO: Feeds and Groups

README.md

What's this for?

This gist demonstrates feed + group context support for Adafruit IO's MQTT service.

Running this demo

On Adafruit IO

1. Create group: "Garage"
2. Create group: "Office"
3. Create a feed, "Signal" and add it to both groups

On your local machine

Get the code:

1. Git clone this gist: git clone https://gist.github.com/a04694748ad887d5aa7e644e3292fa81.git io-mqtt-demo
2. Navigate to the new project.
3. Install dependencies with bundle install (installing bundler first, if you haven't yet)

Setup environment variables, run the following commands in your shell or add them to your .profile. NOTE: this step requires MacOS or other linux-friendly shell.

export IO_USERNAME='yourusername'
export IO_KEY='your-adafruit-io-key'

Run the code:

1. Run ruby subscriber.rb
2. In a new terminal, with the environment variables set, run ruby publisher.rb

And watch the data fly!

Gemfile

source 'https://rubygems.org'
gem 'mqtt'

publisher.rb

require_relative './utilities'

topics = [
  "#{USER}/feeds/office.signal",
  "#{USER}/feeds/garage.signal",
]

count = 0
with_client_connection do |client|
  loop do
    count += 1

    # alternate between topics
    topic = topics[count % 2]

    value = count.to_s
    puts "PUBLISH #{topic} \"#{value}\""

    client.publish(topic, value, true)

    # pause 5 seconds between publish events
    sleep 5
  end
end

subscriber.rb

require_relative './utilities'

topics = [
  "#{USER}/feeds/office.signal",
  "#{USER}/feeds/garage.signal",
  "#{USER}/groups/garage",
  "#{USER}/groups/office",
]

topics.each do |topic|
  puts "SUBSCRIBE #{topic}"
end

subscribing_to(topics) do |topic, message|
  puts "#{topic} >> #{message}"
end

utilities.rb

# A few helper methods for simplifying MQTT connections to Adafruit IO in Ruby scripts.

# gem install mqtt
require 'mqtt'
require 'securerandom'

USER     = ENV['IO_USERNAME']
PASS     = ENV['IO_KEY']

URL      = 'io.adafruit.com'
PROTOCOL = 'mqtts'
PORT     = 8883

def get_client_connection(options={})
  uri = "#{PROTOCOL}://#{USER}:#{PASS}@#{URL}"

  if options[:uri]
    uri = options[:uri]
    options.delete(:uri)
  end

  puts "CONNECT TO #{ uri.gsub(/:[^@]+@/, ':xxxxxxxxxx@') }"
  MQTT::Client.connect(uri, PORT, { ack_timeout: 10, client_id: "io-mqtt-" + SecureRandom.hex(4) }.merge(options))
end

def with_client_connection(options={})
  begin
    yield get_client_connection(options)
  rescue MQTT::ProtocolException => ex
    puts "LOST MQTT CONNECTION IN child PROCESS #{ Process.pid }"

    # disconnected, sleep, reconnect and retry in around 10 seconds
    sleep(8 + rand(5))

    retry
  rescue => ex
    puts "unexpected error in client connection block: #{ ex.message }"
    puts ex.backtrace[0..16]
  end
end

def subscribing_to(topics, client_options={})
  with_client_connection(client_options) do |client|
    client.get(Hash[ topics.map {|t| [t, 1]} ]) do |topic, message|
      yield topic, message
    end
  end
end