MQTT subscribe - Printable Version

Code:

-- Define a function which is called by mqtt_client:handler(),
-- whenever messages are received on the subscribed topics

    function callback(topic, message)
    print("Received: " .. topic .. ": " .. message)
    if (message == "quit") then running = false end
  end

-- Create an MQTT client instance, connect to the MQTT server and
-- subscribe to the topic called "arqueta/humedad"

    MQTT = require("mosquitto")
  MQTT.Utility.set_debug(true)
    mqtt_client = MQTT.client.create("https://test.mosquitto.org", 1883, callback)
    mqtt_client:connect("logicmachine")
    local humedad = mqtt_client:subscribe("arqueta/humedad")
    grp.update('1/1/27', humedad)


-- Continously invoke mqtt_client:handler() to process the MQTT protocol and
-- handle any received messages.  Also, publish a message on topic "test/1"

  running = true

  while (running) do
    mqtt_client:handler()
    mqtt_client:publish("test/1", "test message")
    socket.sleep(1.0)  -- seconds
  end

-- Clean-up by unsubscribing from the topic and closing the MQTT connection

  mqtt_client:unsubscribe({"arqueta/humedad"})
  mqtt_client:destroy()

Code:

if not broker then
  broker = '192.168.1.101'

  function multiply(mult)
    return function(value)
      local num = tonumber(value)
      if num then
        return num * mult
      else
        return value
      end
    end
  end

  -- topic to object map
  mqtt_to_object = {
    ['in/topic1'] = '32/1/1',
    ['in/topic2'] = '32/1/2',
  }

  -- optional topic value conversion function
  mqtt_to_object_conv = {
    ['in/topic1'] = multiply(100),
    ['in/topic2'] = multiply(0.01),
  }

  -- object to topic map
  object_to_mqtt = {
    ['1/1/1'] = 'out/topic1',
    ['1/1/2'] = 'out/topic2',
  }

  datatypes = {}

  grp.sender = 'mq'
  require('socket')

  for addr, _ in pairs(object_to_mqtt) do
    local obj = grp.find(addr)
    if obj then
      datatypes[ addr ] = obj.datatype
    end
  end

  mclient = require('mosquitto').new()

  mclient.ON_CONNECT = function(res, ...)
    log('mqtt connect status', res, ...)

    if res then
      for topic, _ in pairs(mqtt_to_object) do
        mclient:subscribe(topic)
      end
    else
      mclient:disconnect()
    end
  end

  mclient.ON_MESSAGE = function(mid, topic, payload)
    local addr = mqtt_to_object[ topic ]
    if addr then
      local fn = mqtt_to_object_conv[ topic ]

      if fn then
        payload = fn(payload)
      end

      grp.write(addr, payload)
    end
  end

  mclient.ON_DISCONNECT = function(...)
    log('mqtt disconnect', ...)
    mclientfd = nil
  end

  function mconnect()
    local fd

    mclient:connect(broker)
    fd = mclient:socket()

    -- fd ref is valid
    if fd then
      mclientfd = fd
    end
  end

  mconnect()

  function publishvalue(event)
    -- message from us or client is not connected
    if event.sender == 'mq' or not mclientfd then
      return
    end

    local addr = event.dst
    local dpt = datatypes[ addr ]
    local topic = object_to_mqtt[ addr ]

    -- unknown object
    if not dpt or not topic then
      return
    end

    local value = busdatatype.decode(event.datahex, dpt)
    if value ~= nil then
      if type(value) == 'boolean' then
        value = value and 1 or 0
      end

      mclient:publish(topic, tostring(value))
    end
  end

  lbclient = require('localbus').new(1)
  lbclient:sethandler('groupwrite', publishvalue)

  lbclientfd = socket.fdmaskset(lbclient:getfd(), 'r')

  -- run timer every 5 seconds
  timer = require('timerfd').new(5)
  timerfd = socket.fdmaskset(timer:getfd(), 'r')
end

-- mqtt connected
if mclientfd then
  mclientfdset = socket.fdmaskset(mclientfd, mclient:want_write() and 'rw' or 'r')
  res, lbclientstat, timerstat, mclientstat =
      socket.selectfds(10, lbclientfd, timerfd, mclientfdset)
-- mqtt not connected
else
  res, lbclientstat, timerstat =
    socket.selectfds(10, lbclientfd, timerfd)
end

if mclientfd and mclientstat then
  if socket.fdmaskread(mclientstat) then
    mclient:loop_read()
  end

  if socket.fdmaskwrite(mclientstat) then
    mclient:loop_write()
  end
end

if lbclientstat then
  lbclient:step()
end

if timerstat then
  -- clear armed timer
  timer:read()

  if mclientfd then
    mclient:loop_misc()
  else
    mconnect()
  end
end

Code:

local r, e, c = mclient:publish(topic, tostring(value))
log('publish', r, e, c)

Code:

mclient.ON_CONNECT = function(res, ...)
  log('mqtt connect status', res, ...)

  if res then
    mclient:subscribe('100')
  else
    mclient:disconnect()
  end
end

mclient.ON_MESSAGE = function(mid, topic, payload)
  if topic == '100' then
    if payload == "C,1,1,\r\n" then
      grp.write('0/0/1', true)
    elseif payload == "C,1,0,\r\n" then
      grp.write('0/0/1', false)
    elseif payload == "C,2,1,\r\n" then
      grp.write('0/0/2', true)
    elseif payload == "C,2,0,\r\n" then
      grp.write('0/0/2', false) 
    end
  end
end