mqtt & influxdb sample..

mqtt_influxdb_sample.rb

#!/usr/bin/ruby
# -*- encoding: utf-8 -*-
#
# mqtt & influxdb sample...
#
# libraries
#   $ gem install mqtt
#   $ gem install influxdb
#

require 'rubygems'
require 'mqtt'
require 'influxdb'
require 'json'
require 'logger'

# configure
mqtt_params = {
    :remote_host => 'localhost',
    :remote_port => 1883,
    :username    => 'username',
    :password    => 'password',
}

influxdb_params = {
    :host     => 'localhost',
    :username => 'username',
    :password => 'password',
}
database         = 'database_name'

STDOUT.sync = true
log = Logger.new(STDOUT)
log.level = Logger::INFO

influxdb = InfluxDB::Client.new(database, influxdb_params)

loop do
    begin
        MQTT::Client.connect(mqtt_params) do |c|
            log.info("connection start...")
            c.get('#') do |topic, message|
                log.info("recieve mqtt message : topic=#{topic}, message=#{message}")

                # check json format
                json = JSON.parse(message)

                # write
                json[:time] = Time.now.to_i
                influxdb.write_point(topic, json, false, 's')

                log.info("write : database=#{database}, series=#{topic}, values=#{json.to_s}")
            end
        end
    rescue Exception => e
        log.error(e)
    end
    log.info("connection retry...")
    sleep 3
end