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