Instantly share code, notes, and snippets.

@abachman /Gemfile
Last active Nov 2, 2017

Embed
What would you like to do?
MQTT Ruby + Adafruit IO: Feeds and Groups

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!

source 'https://rubygems.org'
gem 'mqtt'
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
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
# 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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment