MQTT handling was the primary goal when Séléné was created : it has a quit complete coverage using SelMQTT object as described bellow.

The full source code of this example is here.

Brk, err = SelMQTT.connect( "tcp://localhost:1883", { reliable=false  } )
if not Brk then
        print( err )
        return
end

SelMQTT.connect() will connect to the broker for which the URL provided as first argument. The second argument is a table of options

• KeepAliveInterval, cleansession, reliable : please have a look on PAHO's documentation
• username, password : for MQTT 3.1 authentification
• OnDisconnect, OnDisconnectTrigger = classback to be called if the broker got disconnected. With OnDisconnectTrigge, the callback is push in task list.

_, err = Brk:subscribe( {
        { topic = "tata/+/truc", func=handle_tata, trigger=update_tata, trigger_once=false, qos=SelMQTT.QoSConst("QoS0") },
        { topic = "toto", trigger=update_toto, trigger_once=true }
} )
if err then
        print( err )
        return
end

Broker's subscribe takes a table as argument, a row by topic to listen to, and with following content

The topic to be listen : this argument is obviously mandatory and may contain MQTT's wildcards.

It's possible to specify the Quality Of Service to apply : QoS0 by default.

Specify the function to be launched immediately, in a separate thread, when a message arrives.

function handle_tata( topic, msg )
        print("Lua received t:'" .. topic .."' m:'".. msg .."'")
        SelShared.set('tata', msg)
        return true
end

As explained in multi-threading discuss, such function has no access to global variables.

If not specified, a default function is called which only create a SelShared with topic as name and containing the message as value. Notez-bien : if you specify yourself a function, this SelShared is not created but if you do it in your function.

If the function return true, trigger function (if specified) is submited to the waiting tasks list. Otherwise, it isn't.

Unlike functions, triggers are not launched immediately at message arrival but are pushed in tasks waiting list and doesn't receive any argument : typically, it retrieves the value to handle from a SelShared.

function update_toto()
        print("Trigger for toto :", SelShared.get('toto'))
end

To avoid duplication in tasks waiting list, use trigger_once argument.

It's possible to associate a watchdog to a topic.

The first step is to create a timer

function exhausted()
        print("No message received after 3 seconds")
end

-- Create the watchdog's timer
wdtimer, err = SelTimer.create { when=3, ifunc=exhausted, clockid=SelTimer.ClockModeConst("CLOCK_MONOTONIC") }

Exhausted() is executed if no data is received within 3 seconds.

Then it is associated to the topic :

_, err = Brk:subscribe( {
 { topic = "toto", trigger=msgreceived, watchdog=wdtimer }
} )

The waiting loop is modified to take in account this timer.

while true do
    local rt = table.pack( Selene.WaitFor(wdtimer) )
    for _,ret in ipairs(rt) do
        if type(ret) == 'function' then
            ret()
         end
     end
end