How to send a respond message with mqtt_as?

[klinwer]

Hello to everyone.
I try to implement a respond message like a sensor value after a request message. It should be a kind of handshake.
Are there any specific examples which I could use as a starting point.

[peterhinch]

The mqtt_as repo has several examples. See, for example, range.py.

[klinwer]

I ran range_ex.py successfully but don't get the point how to send a response message triggered by a received message. Shouldn't be there a seperate thread with a queue in for this?

[beetlegigg]

I've probably not understood what you are asking as the whole point of mqtt is for client programs to publish messages (information or requests?) to another client program that can do something with the information or respond by publishing a response message.

I have a very simple snippet to hand so with a quick tweak I show it here in case its what you are after. The snippet cheats a bit because it starts off by connecting to a broker, but then publishes a mqtt message to which the program has subscribed to. Normally one would expect this published request message to come from another program. But as you can see the program subscribes to the topic of 'test/request'. When it receives such a message (and it will because the program also publishes a message with this topic) then the message callback function check for a message on the subject of 'test/request' and if so calls the function response(). This function then prints a supposes sensor value (set as a global variable for this example) and also publish an mqtt message on the topic of 'test/RESPONSE' with the payload containing the value of the supposed sensor. Maybe this very noddy and cut down mqtt example helps?

# Based on the mqtt_as.py library
# (C) Copyright Peter Hinch 2017-2019.
# Released under the MIT licence.
# and accompanying exaple code

from mqtt_as import MQTTClient, config
import uasyncio as asyncio

import time


# ------------------------------------------------------
# config settings (config dic imported from mqtt_as)
# CHANGE AS REQUIRED

# mqtt server
config['server'] = '10.0.1.141'  # CHANGE FOR YOUR IP

# WiFi
config['ssid'] = 'YOUR SSID'
config['wifi_pw'] = 'YOUR PASSWORD'

# Sensor value for demo purposes
sensor = 23.5

# ------------------------------------------------------

def response():
    # printing the response
    print('the sensor value is', sensor)
    
    # publishing the response     
    asyncio.create_task(client.publish('test/RESPONSE', str(sensor)))   


#-----------------------------------------------------------
# MQTT

# mqtt message received callback
def sub_cb(topic, msg, retained):
    topics = topic.decode("utf-8")
    payload = msg.decode()
    if topics == 'test/request':
        response()
        
    
# mqtt connection subscriptions 
async def conn_han(client):
    #Subscribing to topics
    await client.subscribe('test/request')
 
    
#----------------------------------------------------------------
# main function - called by asyncio.run 
async def main(client):
    try:
        print('trying to connect to MQTT broker')
        await client.connect()
        print('connected to MQTT broker')
    except OSError:
        print('MQTT connection failed.')
    
    while True:
        await client.publish('test/request','')
        await asyncio.sleep(5)
        

#---------------------------------------------------------------
async def wifi_han(state):
    pass


# Define configuration for mqtt_as library
config['subs_cb'] = sub_cb
config['wifi_coro'] = wifi_han
config['connect_coro'] = conn_han
config['clean'] = True

# Set up client
MQTTClient.DEBUG = False  # Optional
client = MQTTClient(config)


try:
    asyncio.run(main(client))
finally:
    client.close()  # Prevent LmacRxBlk:1 errors
    asyncio.new_event_loop()

[klinwer]

Thanks for this snippet. This is what I was looking for. It shows the structure I need.

[peterhinch]

The suggestion from @beetlegigg looks good.

Alternatively using the event-based interface, something like this (adapted from range.py):

async def messages(client):
    async for topic, msg, retained in client.queue:
        asyncio.create_task(client.publish('test/RESPONSE', str(sensor)))

[beetlegigg]

Peter I was aware of your latest spiffing event based interface of a month or so back but I have been amiss in amending my personal mqtt_as snippets. To ensure I use the event interface going forward I have reworked the noddy demo snippet I posted to include in my snippet files and include it here in case it helps the OP to see a comparison to the callback based snippet I showed.

from mqtt_as import MQTTClient, config
import uasyncio as asyncio

# ------------------------------------------------------
# config settings (config dic imported from mqtt_as)
# CHANGE AS REQUIRED

config['ssid'] = 'YOUR SSID'  
config['wifi_pw'] = 'YOUR PASSWORD'
config['server'] = '10.0.1.141' # CHANGE FOR YOUR IP
# ------------------------------------------------------

sensor = 23.44

# ------------------------------------------------------

# mqtt messages received
async def messages(client):  # Respond to incoming messages
    async for topic, msg, retained in client.queue:
        if topic == 'test/request':
             # printing the response
            print('the sensor value is', sensor)
    
            # publishing the response     
            asyncio.create_task(client.publish('test/RESPONSE', str(sensor)))
        else:
            # print out all other mqtt messages are received
            print((topic, msg, retained))


# mqtt connection subscriptions 
async def up(client):  # Respond to connectivity being (re)established
    while True:
        await client.up.wait()  # Wait on an Event
        client.up.clear()
        await client.subscribe('test/request', 1)  # renew subscriptions

#----------------------------------------------------------------
# main function - called by asyncio.run 
async def main(client):
    await client.connect()
    for coroutine in (up, messages):
        asyncio.create_task(coroutine(client))
    n = 0
    while True:
        await asyncio.sleep(5)
        print('publish', n)
        
        # for testing publish a message on the topic of 'test/request'
        await client.publish('test/request', '')

        # If WiFi is down the following will pause for the duration.
        await client.publish('result', '{}'.format(n), qos = 1)
        n += 1

# -------------------------------------------------------
# configure for Event based interface (instead of callbacks)
config["queue_len"] = 1  # Use event interface with default queue size

# Set up mqtt client
MQTTClient.DEBUG = True  # Optional: print diagnostic messages
client = MQTTClient(config)

try:
    asyncio.run(main(client))
finally:
    client.close()  # Prevent LmacRxBlk:1 errors

[klinwer]

The sticking point for me was "if topic == 'test/request" because this shows that I have to use different topics instead of one topic with all options and routines.

[beetlegigg]

@klinwer, good to hear you are now unstuck :-) . Its probably obvious, but a couple of things maybe worth reminding for one starting out to use mqtt.

First is to remember that only mqtt messages that are subscribed to will be received by your program so you don't necessarily need the if topic ==. In the example snippet only one message topic was subscribed to so in that case the 'if topic' is superfluous. Of course for multiple subscriptions it does allow different actions to be taken depending on the message topic.

Also my example did not receive a message payload with the test/request topic (as none was published), but the message payload could also signal a particular action. The payload could also be formatted, perhaps as a dictionary of values etc. For example it could contain a json formatted message where different actions could be taken depending on the content of a particular item in the structure. Likewise the test/RESPONSE message with the payload of the sensor value could have been sending a message that contained a whole bunch of different sensor readings etc. and the received program may take some action based on just one of the payload items. (e.g. take action to switch on heater if temperature is too low, but just display both the temperature the humidity value as received in the payload.)

And as regards the @peterhinch mqtt_as, as I understand it, both event and callback options will work fine, but the preference is to go with the event based option for new programs. I am cogitating if I should change any of my existing programs as I presume the event based option is more efficient and works better with asynchronous programs and I will probably learn something in the exercise. (updates to Peters programs have been keeping me on my toes over the last few months :-) )

You caught me drinking a cuppa so forgive the probably unwarranted over voluble response.

[peterhinch]

I presume the event based option is more efficient

In terms of RAM use it's slightly less efficient. However I think it leads to clearer, more easily understood code. In the specific case of MQTT, the event based approach has predictable behaviour if a set of messages arrive quickly: they are queued and processed in order. Of course the callback occurs in order, but if the callback launches a task, multiple instances of that task may run concurrently. This uses resources and (potentially) confuses the coder :)

I have come to the conclusion that callbacks are rather against the spirit of uasyncio given that we have a very efficient Event class (a task waiting on an Event is put on a queue and uses no CPU time).

On extremely RAM constrained devices like ESP8266 I might stick with callbacks, but I do prefer Events...

[klinwer]

Your hints are really helpfull. I will test the event based option also. The other solution is already working fine on an ESP32c3. I also have to incooperate the new topic structure on server side.
Regards Werner

[klinwer]

The update to the event based option is working. 👍 Thanks a lot.