rp2da2.mqtt.mqtt_client.MQTTClient

class documentation

class MQTTClient(Device):

Constructor: MQTTClient()

MQTT Client

This is a simplified version of an MQTT Client. Quality of Service 2 is not supported.

Subscriptions are pre-defined and set up on connection. Dynamic modifications to subscriptions not supported.

It is a singleton.

The client is started by calling the start method with a list of MQTTSubscription objects. The client will connect to the MQTT broker and subscribe to the topics defined in the subscription list. The client will then listen for incoming messages and handle them according to the subscription list.

asyncio.Streams are used for reading and writing.

Class Method	`get_instance`	Return the singleton instance
Method	`__init__`	Construct the MQTT Client
Async Method	`publish`	MQTT Publish
Async Method	`run`	MQTT Client Run
Constant	`DUP`	MQTT duplicate flag
Constant	`ERR_CONACK`	invalid connection acknowledgement
Constant	`ERR_LEN`	ill formed length field
Constant	`ERR_OSR`	Operating System Error on read or write
Constant	`ERR_P_TYPE`	unrecognised packet type
Constant	`ERR_PINGRESP`	invalid ping response
Constant	`ERR_PUBACK`	invalid publish acknowledgement
Constant	`ERR_SUBACK`	invalid subscription acknowledgement
Constant	`ERR_UTF8`	invalid UTF8 character
Constant	`M_CLOSED`	no connection
Constant	`M_CONNECTED`	status connected (and subscribed)
Constant	`M_CONNECTING`	new connection - waiting for CONNACK
Constant	`M_PING_SENT`	connected - waiting for PINGRESP
Constant	`M_REJECTED`	connection rejected by broker (it will terminate it!)
Constant	`M_SUBSCRIBING`	new connection - waiting for SUBACK
Constant	`PROTOCOL_LVL`	MQTT Protocol Level 4 (MQTT version 3.1.1)
Constant	`QOS0`	Quality of service 0
Constant	`QOS1`	Quality of service 1
Constant	`QOS_MASK`	mask for QoS bits
Constant	`RETAIN`	MQTT retain flag
Instance Variable	`errors`	counts of errors by type
Property	`broker`	The MQTT Broker ID.
Async Method	`_close`	We only close the stream as part of error recovery
Async Method	`_handle_connack`	Undocumented
Async Method	`_handle_pingresp`	Undocumented
Async Method	`_handle_puback`	PUBACK received
Async Method	`_handle_publish`	PUBLISH received
Async Method	`_handle_suback`	Undocumented
Async Method	`_pinger`	Coroutine to send a ping
Async Method	`_puback`	Return PUBACK after receiving PUBLISH
Async Method	`_re_open`	Open or Re-open the MQTT connection
Method	`_set_state`	Undocumented
Async Method	`_subscribe`	Undocumented
Constant	`_RESP_HANDLER`	Undocumented
Class Variable	`_mqtt_client`	Undocumented
Instance Variable	`_client_id`	Undocumented
Instance Variable	`_clientPidTx`	Undocumented
Instance Variable	`_con_params`	Undocumented
Instance Variable	`_ping_deferred`	Undocumented
Instance Variable	`_ping_task`	Undocumented
Instance Variable	`_reader`	Undocumented
Instance Variable	`_state`	Undocumented
Instance Variable	`_subscription_list`	Undocumented
Instance Variable	`_wifi`	Undocumented
Instance Variable	`_writer`	Undocumented

@classmethod

def get_instance(cls): ¶

Return the singleton instance

The singleton is created on the first call.

def __init__(self): ¶

Construct the MQTT Client

Getting a reference to the Wi-Fi singleton will instantiate it if not already done.

async def publish(self, topic, payload, retain=True, qos=QOS1): ¶

MQTT Publish

At the moment we will assume no duplicates will be sent it only makes sense to send duplicates if clean session not used. I.e. sessions may span connections - therefore despite use of QoS one there's no need to save the message.

TODO Trap 'no free pid error'

Parameters
topic	MQTT topic
payload	the payload message to be sent
retain	boolean
qos	Quality of Service

async def run(self, subscription_list): ¶

MQTT Client Run

This coroutine runs the MQTT client. It does not return.

Parameters
subscription_list	a list of MQTTSubscripion objects to subscribe to

DUP = ¶

MQTT duplicate flag

Value

const(8)

ERR_CONACK = ¶

invalid connection acknowledgement

Value

const(3)

ERR_LEN = ¶

ill formed length field

Value

const(1)

ERR_OSR = ¶

Operating System Error on read or write

Value

const(8)

ERR_P_TYPE = ¶

unrecognised packet type

Value

const(2)

ERR_PINGRESP = ¶

invalid ping response

Value

const(4)

ERR_PUBACK = ¶

invalid publish acknowledgement

Value

const(5)

ERR_SUBACK = ¶

invalid subscription acknowledgement

Value

const(6)

ERR_UTF8 = ¶

invalid UTF8 character

Value

const(7)

M_CLOSED = ¶

no connection

Value

const(0)

M_CONNECTED = ¶

status connected (and subscribed)

Value

const(3)

M_CONNECTING = ¶

new connection - waiting for CONNACK

Value

const(1)

M_PING_SENT = ¶

connected - waiting for PINGRESP

Value

const(4)

M_REJECTED = ¶

connection rejected by broker (it will terminate it!)

Value

const(5)

M_SUBSCRIBING = ¶

new connection - waiting for SUBACK

Value

const(2)

PROTOCOL_LVL = ¶

MQTT Protocol Level 4 (MQTT version 3.1.1)

Value

const(4)

QOS0 = ¶

Quality of service 0

Value

const(0)

QOS1 = ¶

Quality of service 1

Value

const(1)

QOS_MASK = ¶

mask for QoS bits

Value

const(6)

RETAIN = ¶

MQTT retain flag

Value

const(1)

errors = ¶

counts of errors by type

@property

broker = ¶

The MQTT Broker ID.

This returns the MQTT Broker ID as a string.

async def _close(self): ¶

We only close the stream as part of error recovery

async def _handle_connack(self, pf, packet): ¶

Undocumented

async def _handle_pingresp(self, pf, packet): ¶

Undocumented

async def _handle_puback(self, pf, packet): ¶

PUBACK received

Acknowledgement of PUBLISH QoS1

async def _handle_publish(self, pf, packet): ¶

PUBLISH received

async def _handle_suback(self, pf, packet): ¶

Undocumented

async def _pinger(self): ¶

Coroutine to send a ping

The ping will be sent when the ping timer expires. The coroutine will close the connection and terminate if the previous ping has not been acknowledged. The coroutine will exit if no longer connected.

async def _puback(self, pid): ¶

Return PUBACK after receiving PUBLISH

async def _re_open(self): ¶

Open or Re-open the MQTT connection

def _set_state(self, state): ¶

Undocumented

async def _subscribe(self): ¶

Undocumented

_RESP_HANDLER = ¶

Undocumented

Value

{_CONNACK: _handle_connack,
 _PINGRESP: _handle_pingresp,
 _PUBACK: _handle_puback,
 _SUBACK: _handle_suback,
 _PUBLISH: _handle_publish}

_mqtt_client = ¶

Undocumented

_client_id = ¶

Undocumented

_clientPidTx = ¶

Undocumented

_con_params = ¶

Undocumented

_ping_deferred = ¶

Undocumented

_ping_task = ¶

Undocumented

_reader = ¶

Undocumented

_state = ¶

Undocumented

_subscription_list = ¶

Undocumented

_wifi = ¶

Undocumented

_writer = ¶

Undocumented