Git-Mirrors/Sideband
1
0
Fork 0
mirror of https://github.com/markqvist/Sideband.git synced 2025-02-11 04:09:03 -05:00
Code Issues Actions Packages Projects Releases Wiki Activity

Added MQTT library

Browse Source
This commit is contained in:
Mark Qvist 2025-01-22 22:31:16 +01:00
parent 5def619930
commit 9bb4f3cc8b
11 changed files with 6587 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
sbapp/mqtt/__init__.py Normal file
View File

@ -0,0 +1,5 @@
__version__ = "2.1.1.dev0"
class MQTTException(Exception):
pass

5004
sbapp/mqtt/client.py Normal file
View File

File diff suppressed because it is too large Load Diff

113
sbapp/mqtt/enums.py Normal file
View File

@ -0,0 +1,113 @@
import enum
class MQTTErrorCode(enum.IntEnum):
MQTT_ERR_AGAIN = -1
MQTT_ERR_SUCCESS = 0
MQTT_ERR_NOMEM = 1
MQTT_ERR_PROTOCOL = 2
MQTT_ERR_INVAL = 3
MQTT_ERR_NO_CONN = 4
MQTT_ERR_CONN_REFUSED = 5
MQTT_ERR_NOT_FOUND = 6
MQTT_ERR_CONN_LOST = 7
MQTT_ERR_TLS = 8
MQTT_ERR_PAYLOAD_SIZE = 9
MQTT_ERR_NOT_SUPPORTED = 10
MQTT_ERR_AUTH = 11
MQTT_ERR_ACL_DENIED = 12
MQTT_ERR_UNKNOWN = 13
MQTT_ERR_ERRNO = 14
MQTT_ERR_QUEUE_SIZE = 15
MQTT_ERR_KEEPALIVE = 16
class MQTTProtocolVersion(enum.IntEnum):
MQTTv31 = 3
MQTTv311 = 4
MQTTv5 = 5
class CallbackAPIVersion(enum.Enum):
"""Defined the arguments passed to all user-callback.
See each callbacks for details: `on_connect`, `on_connect_fail`, `on_disconnect`, `on_message`, `on_publish`,
`on_subscribe`, `on_unsubscribe`, `on_log`, `on_socket_open`, `on_socket_close`,
`on_socket_register_write`, `on_socket_unregister_write`
"""
VERSION1 = 1
"""The version used with paho-mqtt 1.x before introducing CallbackAPIVersion.
This version had different arguments depending if MQTTv5 or MQTTv3 was used. `Properties` & `ReasonCode` were missing
on some callback (apply only to MQTTv5).
This version is deprecated and will be removed in version 3.0.
"""
VERSION2 = 2
""" This version fix some of the shortcoming of previous version.
Callback have the same signature if using MQTTv5 or MQTTv3. `ReasonCode` are used in MQTTv3.
"""
class MessageType(enum.IntEnum):
CONNECT = 0x10
CONNACK = 0x20
PUBLISH = 0x30
PUBACK = 0x40
PUBREC = 0x50
PUBREL = 0x60
PUBCOMP = 0x70
SUBSCRIBE = 0x80
SUBACK = 0x90
UNSUBSCRIBE = 0xA0
UNSUBACK = 0xB0
PINGREQ = 0xC0
PINGRESP = 0xD0
DISCONNECT = 0xE0
AUTH = 0xF0
class LogLevel(enum.IntEnum):
MQTT_LOG_INFO = 0x01
MQTT_LOG_NOTICE = 0x02
MQTT_LOG_WARNING = 0x04
MQTT_LOG_ERR = 0x08
MQTT_LOG_DEBUG = 0x10
class ConnackCode(enum.IntEnum):
CONNACK_ACCEPTED = 0
CONNACK_REFUSED_PROTOCOL_VERSION = 1
CONNACK_REFUSED_IDENTIFIER_REJECTED = 2
CONNACK_REFUSED_SERVER_UNAVAILABLE = 3
CONNACK_REFUSED_BAD_USERNAME_PASSWORD = 4
CONNACK_REFUSED_NOT_AUTHORIZED = 5
class _ConnectionState(enum.Enum):
MQTT_CS_NEW = enum.auto()
MQTT_CS_CONNECT_ASYNC = enum.auto()
MQTT_CS_CONNECTING = enum.auto()
MQTT_CS_CONNECTED = enum.auto()
MQTT_CS_CONNECTION_LOST = enum.auto()
MQTT_CS_DISCONNECTING = enum.auto()
MQTT_CS_DISCONNECTED = enum.auto()
class MessageState(enum.IntEnum):
MQTT_MS_INVALID = 0
MQTT_MS_PUBLISH = 1
MQTT_MS_WAIT_FOR_PUBACK = 2
MQTT_MS_WAIT_FOR_PUBREC = 3
MQTT_MS_RESEND_PUBREL = 4
MQTT_MS_WAIT_FOR_PUBREL = 5
MQTT_MS_RESEND_PUBCOMP = 6
MQTT_MS_WAIT_FOR_PUBCOMP = 7
MQTT_MS_SEND_PUBREC = 8
MQTT_MS_QUEUED = 9
class PahoClientMode(enum.IntEnum):
MQTT_CLIENT = 0
MQTT_BRIDGE = 1

78
sbapp/mqtt/matcher.py Normal file
View File

@ -0,0 +1,78 @@
class MQTTMatcher:
"""Intended to manage topic filters including wildcards.
Internally, MQTTMatcher use a prefix tree (trie) to store
values associated with filters, and has an iter_match()
method to iterate efficiently over all filters that match
some topic name."""
class Node:
__slots__ = '_children', '_content'
def __init__(self):
self._children = {}
self._content = None
def __init__(self):
self._root = self.Node()
def __setitem__(self, key, value):
"""Add a topic filter :key to the prefix tree
and associate it to :value"""
node = self._root
for sym in key.split('/'):
node = node._children.setdefault(sym, self.Node())
node._content = value
def __getitem__(self, key):
"""Retrieve the value associated with some topic filter :key"""
try:
node = self._root
for sym in key.split('/'):
node = node._children[sym]
if node._content is None:
raise KeyError(key)
return node._content
except KeyError as ke:
raise KeyError(key) from ke
def __delitem__(self, key):
"""Delete the value associated with some topic filter :key"""
lst = []
try:
parent, node = None, self._root
for k in key.split('/'):
parent, node = node, node._children[k]
lst.append((parent, k, node))
# TODO
node._content = None
except KeyError as ke:
raise KeyError(key) from ke
else: # cleanup
for parent, k, node in reversed(lst):
if node._children or node._content is not None:
break
del parent._children[k]
def iter_match(self, topic):
"""Return an iterator on all values associated with filters
that match the :topic"""
lst = topic.split('/')
normal = not topic.startswith('$')
def rec(node, i=0):
if i == len(lst):
if node._content is not None:
yield node._content
else:
part = lst[i]
if part in node._children:
for content in rec(node._children[part], i + 1):
yield content
if '+' in node._children and (normal or i > 0):
for content in rec(node._children['+'], i + 1):
yield content
if '#' in node._children and (normal or i > 0):
content = node._children['#']._content
if content is not None:
yield content
return rec(self._root)

43
sbapp/mqtt/packettypes.py Normal file
View File

@ -0,0 +1,43 @@
"""
*******************************************************************
Copyright (c) 2017, 2019 IBM Corp.
All rights reserved. This program and the accompanying materials
are made available under the terms of the Eclipse Public License v2.0
and Eclipse Distribution License v1.0 which accompany this distribution.
The Eclipse Public License is available at
http://www.eclipse.org/legal/epl-v20.html
and the Eclipse Distribution License is available at
http://www.eclipse.org/org/documents/edl-v10.php.
Contributors:
Ian Craggs - initial implementation and/or documentation
*******************************************************************
"""
class PacketTypes:
"""
Packet types class. Includes the AUTH packet for MQTT v5.0.
Holds constants for each packet type such as PacketTypes.PUBLISH
and packet name strings: PacketTypes.Names[PacketTypes.PUBLISH].
"""
indexes = range(1, 16)
# Packet types
CONNECT, CONNACK, PUBLISH, PUBACK, PUBREC, PUBREL, \
PUBCOMP, SUBSCRIBE, SUBACK, UNSUBSCRIBE, UNSUBACK, \
PINGREQ, PINGRESP, DISCONNECT, AUTH = indexes
# Dummy packet type for properties use - will delay only applies to will
WILLMESSAGE = 99
Names = ( "reserved", \
"Connect", "Connack", "Publish", "Puback", "Pubrec", "Pubrel", \
"Pubcomp", "Subscribe", "Suback", "Unsubscribe", "Unsuback", \
"Pingreq", "Pingresp", "Disconnect", "Auth")

421
sbapp/mqtt/properties.py Normal file
View File

@ -0,0 +1,421 @@
# *******************************************************************
# Copyright (c) 2017, 2019 IBM Corp.
#
# All rights reserved. This program and the accompanying materials
# are made available under the terms of the Eclipse Public License v2.0
# and Eclipse Distribution License v1.0 which accompany this distribution.
#
# The Eclipse Public License is available at
# http://www.eclipse.org/legal/epl-v20.html
# and the Eclipse Distribution License is available at
# http://www.eclipse.org/org/documents/edl-v10.php.
#
# Contributors:
# Ian Craggs - initial implementation and/or documentation
# *******************************************************************
import struct
from .packettypes import PacketTypes
class MQTTException(Exception):
pass
class MalformedPacket(MQTTException):
pass
def writeInt16(length):
# serialize a 16 bit integer to network format
return bytearray(struct.pack("!H", length))
def readInt16(buf):
# deserialize a 16 bit integer from network format
return struct.unpack("!H", buf[:2])[0]
def writeInt32(length):
# serialize a 32 bit integer to network format
return bytearray(struct.pack("!L", length))
def readInt32(buf):
# deserialize a 32 bit integer from network format
return struct.unpack("!L", buf[:4])[0]
def writeUTF(data):
# data could be a string, or bytes. If string, encode into bytes with utf-8
if not isinstance(data, bytes):
data = bytes(data, "utf-8")
return writeInt16(len(data)) + data
def readUTF(buffer, maxlen):
if maxlen >= 2:
length = readInt16(buffer)
else:
raise MalformedPacket("Not enough data to read string length")
maxlen -= 2
if length > maxlen:
raise MalformedPacket("Length delimited string too long")
buf = buffer[2:2+length].decode("utf-8")
# look for chars which are invalid for MQTT
for c in buf: # look for D800-DFFF in the UTF string
ord_c = ord(c)
if ord_c >= 0xD800 and ord_c <= 0xDFFF:
raise MalformedPacket("[MQTT-1.5.4-1] D800-DFFF found in UTF-8 data")
if ord_c == 0x00: # look for null in the UTF string
raise MalformedPacket("[MQTT-1.5.4-2] Null found in UTF-8 data")
if ord_c == 0xFEFF:
raise MalformedPacket("[MQTT-1.5.4-3] U+FEFF in UTF-8 data")
return buf, length+2
def writeBytes(buffer):
return writeInt16(len(buffer)) + buffer
def readBytes(buffer):
length = readInt16(buffer)
return buffer[2:2+length], length+2
class VariableByteIntegers: # Variable Byte Integer
"""
MQTT variable byte integer helper class. Used
in several places in MQTT v5.0 properties.
"""
@staticmethod
def encode(x):
"""
Convert an integer 0 <= x <= 268435455 into multi-byte format.
Returns the buffer converted from the integer.
"""
if not 0 <= x <= 268435455:
raise ValueError(f"Value {x!r} must be in range 0-268435455")
buffer = b''
while 1:
digit = x % 128
x //= 128
if x > 0:
digit |= 0x80
buffer += bytes([digit])
if x == 0:
break
return buffer
@staticmethod
def decode(buffer):
"""
Get the value of a multi-byte integer from a buffer
Return the value, and the number of bytes used.
[MQTT-1.5.5-1] the encoded value MUST use the minimum number of bytes necessary to represent the value
"""
multiplier = 1
value = 0
bytes = 0
while 1:
bytes += 1
digit = buffer[0]
buffer = buffer[1:]
value += (digit & 127) * multiplier
if digit & 128 == 0:
break
multiplier *= 128
return (value, bytes)
class Properties:
"""MQTT v5.0 properties class.
See Properties.names for a list of accepted property names along with their numeric values.
See Properties.properties for the data type of each property.
Example of use::
publish_properties = Properties(PacketTypes.PUBLISH)
publish_properties.UserProperty = ("a", "2")
publish_properties.UserProperty = ("c", "3")
First the object is created with packet type as argument, no properties will be present at
this point. Then properties are added as attributes, the name of which is the string property
name without the spaces.
"""
def __init__(self, packetType):
self.packetType = packetType
self.types = ["Byte", "Two Byte Integer", "Four Byte Integer", "Variable Byte Integer",
"Binary Data", "UTF-8 Encoded String", "UTF-8 String Pair"]
self.names = {
"Payload Format Indicator": 1,
"Message Expiry Interval": 2,
"Content Type": 3,
"Response Topic": 8,
"Correlation Data": 9,
"Subscription Identifier": 11,
"Session Expiry Interval": 17,
"Assigned Client Identifier": 18,
"Server Keep Alive": 19,
"Authentication Method": 21,
"Authentication Data": 22,
"Request Problem Information": 23,
"Will Delay Interval": 24,
"Request Response Information": 25,
"Response Information": 26,
"Server Reference": 28,
"Reason String": 31,
"Receive Maximum": 33,
"Topic Alias Maximum": 34,
"Topic Alias": 35,
"Maximum QoS": 36,
"Retain Available": 37,
"User Property": 38,
"Maximum Packet Size": 39,
"Wildcard Subscription Available": 40,
"Subscription Identifier Available": 41,
"Shared Subscription Available": 42
}
self.properties = {
# id: type, packets
# payload format indicator
1: (self.types.index("Byte"), [PacketTypes.PUBLISH, PacketTypes.WILLMESSAGE]),
2: (self.types.index("Four Byte Integer"), [PacketTypes.PUBLISH, PacketTypes.WILLMESSAGE]),
3: (self.types.index("UTF-8 Encoded String"), [PacketTypes.PUBLISH, PacketTypes.WILLMESSAGE]),
8: (self.types.index("UTF-8 Encoded String"), [PacketTypes.PUBLISH, PacketTypes.WILLMESSAGE]),
9: (self.types.index("Binary Data"), [PacketTypes.PUBLISH, PacketTypes.WILLMESSAGE]),
11: (self.types.index("Variable Byte Integer"),
[PacketTypes.PUBLISH, PacketTypes.SUBSCRIBE]),
17: (self.types.index("Four Byte Integer"),
[PacketTypes.CONNECT, PacketTypes.CONNACK, PacketTypes.DISCONNECT]),
18: (self.types.index("UTF-8 Encoded String"), [PacketTypes.CONNACK]),
19: (self.types.index("Two Byte Integer"), [PacketTypes.CONNACK]),
21: (self.types.index("UTF-8 Encoded String"),
[PacketTypes.CONNECT, PacketTypes.CONNACK, PacketTypes.AUTH]),
22: (self.types.index("Binary Data"),
[PacketTypes.CONNECT, PacketTypes.CONNACK, PacketTypes.AUTH]),
23: (self.types.index("Byte"),
[PacketTypes.CONNECT]),
24: (self.types.index("Four Byte Integer"), [PacketTypes.WILLMESSAGE]),
25: (self.types.index("Byte"), [PacketTypes.CONNECT]),
26: (self.types.index("UTF-8 Encoded String"), [PacketTypes.CONNACK]),
28: (self.types.index("UTF-8 Encoded String"),
[PacketTypes.CONNACK, PacketTypes.DISCONNECT]),
31: (self.types.index("UTF-8 Encoded String"),
[PacketTypes.CONNACK, PacketTypes.PUBACK, PacketTypes.PUBREC,
PacketTypes.PUBREL, PacketTypes.PUBCOMP, PacketTypes.SUBACK,
PacketTypes.UNSUBACK, PacketTypes.DISCONNECT, PacketTypes.AUTH]),
33: (self.types.index("Two Byte Integer"),
[PacketTypes.CONNECT, PacketTypes.CONNACK]),
34: (self.types.index("Two Byte Integer"),
[PacketTypes.CONNECT, PacketTypes.CONNACK]),
35: (self.types.index("Two Byte Integer"), [PacketTypes.PUBLISH]),
36: (self.types.index("Byte"), [PacketTypes.CONNACK]),
37: (self.types.index("Byte"), [PacketTypes.CONNACK]),
38: (self.types.index("UTF-8 String Pair"),
[PacketTypes.CONNECT, PacketTypes.CONNACK,
PacketTypes.PUBLISH, PacketTypes.PUBACK,
PacketTypes.PUBREC, PacketTypes.PUBREL, PacketTypes.PUBCOMP,
PacketTypes.SUBSCRIBE, PacketTypes.SUBACK,
PacketTypes.UNSUBSCRIBE, PacketTypes.UNSUBACK,
PacketTypes.DISCONNECT, PacketTypes.AUTH, PacketTypes.WILLMESSAGE]),
39: (self.types.index("Four Byte Integer"),
[PacketTypes.CONNECT, PacketTypes.CONNACK]),
40: (self.types.index("Byte"), [PacketTypes.CONNACK]),
41: (self.types.index("Byte"), [PacketTypes.CONNACK]),
42: (self.types.index("Byte"), [PacketTypes.CONNACK]),
}
def allowsMultiple(self, compressedName):
return self.getIdentFromName(compressedName) in [11, 38]
def getIdentFromName(self, compressedName):
# return the identifier corresponding to the property name
result = -1
for name in self.names.keys():
if compressedName == name.replace(' ', ''):
result = self.names[name]
break
return result
def __setattr__(self, name, value):
name = name.replace(' ', '')
privateVars = ["packetType", "types", "names", "properties"]
if name in privateVars:
object.__setattr__(self, name, value)
else:
# the name could have spaces in, or not. Remove spaces before assignment
if name not in [aname.replace(' ', '') for aname in self.names.keys()]:
raise MQTTException(
f"Property name must be one of {self.names.keys()}")
# check that this attribute applies to the packet type
if self.packetType not in self.properties[self.getIdentFromName(name)][1]:
raise MQTTException(f"Property {name} does not apply to packet type {PacketTypes.Names[self.packetType]}")
# Check for forbidden values
if not isinstance(value, list):
if name in ["ReceiveMaximum", "TopicAlias"] \
and (value < 1 or value > 65535):
raise MQTTException(f"{name} property value must be in the range 1-65535")
elif name in ["TopicAliasMaximum"] \
and (value < 0 or value > 65535):
raise MQTTException(f"{name} property value must be in the range 0-65535")
elif name in ["MaximumPacketSize", "SubscriptionIdentifier"] \
and (value < 1 or value > 268435455):
raise MQTTException(f"{name} property value must be in the range 1-268435455")
elif name in ["RequestResponseInformation", "RequestProblemInformation", "PayloadFormatIndicator"] \
and (value != 0 and value != 1):
raise MQTTException(
f"{name} property value must be 0 or 1")
if self.allowsMultiple(name):
if not isinstance(value, list):
value = [value]
if hasattr(self, name):
value = object.__getattribute__(self, name) + value
object.__setattr__(self, name, value)
def __str__(self):
buffer = "["
first = True
for name in self.names.keys():
compressedName = name.replace(' ', '')
if hasattr(self, compressedName):
if not first:
buffer += ", "
buffer += f"{compressedName} : {getattr(self, compressedName)}"
first = False
buffer += "]"
return buffer
def json(self):
data = {}
for name in self.names.keys():
compressedName = name.replace(' ', '')
if hasattr(self, compressedName):
val = getattr(self, compressedName)
if compressedName == 'CorrelationData' and isinstance(val, bytes):
data[compressedName] = val.hex()
else:
data[compressedName] = val
return data
def isEmpty(self):
rc = True
for name in self.names.keys():
compressedName = name.replace(' ', '')
if hasattr(self, compressedName):
rc = False
break
return rc
def clear(self):
for name in self.names.keys():
compressedName = name.replace(' ', '')
if hasattr(self, compressedName):
delattr(self, compressedName)
def writeProperty(self, identifier, type, value):
buffer = b""
buffer += VariableByteIntegers.encode(identifier) # identifier
if type == self.types.index("Byte"): # value
buffer += bytes([value])
elif type == self.types.index("Two Byte Integer"):
buffer += writeInt16(value)
elif type == self.types.index("Four Byte Integer"):
buffer += writeInt32(value)
elif type == self.types.index("Variable Byte Integer"):
buffer += VariableByteIntegers.encode(value)
elif type == self.types.index("Binary Data"):
buffer += writeBytes(value)
elif type == self.types.index("UTF-8 Encoded String"):
buffer += writeUTF(value)
elif type == self.types.index("UTF-8 String Pair"):
buffer += writeUTF(value[0]) + writeUTF(value[1])
return buffer
def pack(self):
# serialize properties into buffer for sending over network
buffer = b""
for name in self.names.keys():
compressedName = name.replace(' ', '')
if hasattr(self, compressedName):
identifier = self.getIdentFromName(compressedName)
attr_type = self.properties[identifier][0]
if self.allowsMultiple(compressedName):
for prop in getattr(self, compressedName):
buffer += self.writeProperty(identifier,
attr_type, prop)
else:
buffer += self.writeProperty(identifier, attr_type,
getattr(self, compressedName))
return VariableByteIntegers.encode(len(buffer)) + buffer
def readProperty(self, buffer, type, propslen):
if type == self.types.index("Byte"):
value = buffer[0]
valuelen = 1
elif type == self.types.index("Two Byte Integer"):
value = readInt16(buffer)
valuelen = 2
elif type == self.types.index("Four Byte Integer"):
value = readInt32(buffer)
valuelen = 4
elif type == self.types.index("Variable Byte Integer"):
value, valuelen = VariableByteIntegers.decode(buffer)
elif type == self.types.index("Binary Data"):
value, valuelen = readBytes(buffer)
elif type == self.types.index("UTF-8 Encoded String"):
value, valuelen = readUTF(buffer, propslen)
elif type == self.types.index("UTF-8 String Pair"):
value, valuelen = readUTF(buffer, propslen)
buffer = buffer[valuelen:] # strip the bytes used by the value
value1, valuelen1 = readUTF(buffer, propslen - valuelen)
value = (value, value1)
valuelen += valuelen1
return value, valuelen
def getNameFromIdent(self, identifier):
rc = None
for name in self.names:
if self.names[name] == identifier:
rc = name
return rc
def unpack(self, buffer):
self.clear()
# deserialize properties into attributes from buffer received from network
propslen, VBIlen = VariableByteIntegers.decode(buffer)
buffer = buffer[VBIlen:] # strip the bytes used by the VBI
propslenleft = propslen
while propslenleft > 0: # properties length is 0 if there are none
identifier, VBIlen2 = VariableByteIntegers.decode(
buffer) # property identifier
buffer = buffer[VBIlen2:] # strip the bytes used by the VBI
propslenleft -= VBIlen2
attr_type = self.properties[identifier][0]
value, valuelen = self.readProperty(
buffer, attr_type, propslenleft)
buffer = buffer[valuelen:] # strip the bytes used by the value
propslenleft -= valuelen
propname = self.getNameFromIdent(identifier)
compressedName = propname.replace(' ', '')
if not self.allowsMultiple(compressedName) and hasattr(self, compressedName):
raise MQTTException(
f"Property '{property}' must not exist more than once")
setattr(self, propname, value)
return self, propslen + VBIlen

306
sbapp/mqtt/publish.py Normal file
View File

@ -0,0 +1,306 @@
# Copyright (c) 2014 Roger Light <roger@atchoo.org>
#
# All rights reserved. This program and the accompanying materials
# are made available under the terms of the Eclipse Public License v2.0
# and Eclipse Distribution License v1.0 which accompany this distribution.
#
# The Eclipse Public License is available at
# http://www.eclipse.org/legal/epl-v20.html
# and the Eclipse Distribution License is available at
# http://www.eclipse.org/org/documents/edl-v10.php.
#
# Contributors:
# Roger Light - initial API and implementation
"""
This module provides some helper functions to allow straightforward publishing
of messages in a one-shot manner. In other words, they are useful for the
situation where you have a single/multiple messages you want to publish to a
broker, then disconnect and nothing else is required.
"""
from __future__ import annotations
import collections
from collections.abc import Iterable
from typing import TYPE_CHECKING, Any, List, Tuple, Union
from paho.mqtt.enums import CallbackAPIVersion, MQTTProtocolVersion
from paho.mqtt.properties import Properties
from paho.mqtt.reasoncodes import ReasonCode
from .. import mqtt
from . import client as paho
if TYPE_CHECKING:
try:
from typing import NotRequired, Required, TypedDict # type: ignore
except ImportError:
from typing_extensions import NotRequired, Required, TypedDict
try:
from typing import Literal
except ImportError:
from typing_extensions import Literal # type: ignore
class AuthParameter(TypedDict, total=False):
username: Required[str]
password: NotRequired[str]
class TLSParameter(TypedDict, total=False):
ca_certs: Required[str]
certfile: NotRequired[str]
keyfile: NotRequired[str]
tls_version: NotRequired[int]
ciphers: NotRequired[str]
insecure: NotRequired[bool]
class MessageDict(TypedDict, total=False):
topic: Required[str]
payload: NotRequired[paho.PayloadType]
qos: NotRequired[int]
retain: NotRequired[bool]
MessageTuple = Tuple[str, paho.PayloadType, int, bool]
MessagesList = List[Union[MessageDict, MessageTuple]]
def _do_publish(client: paho.Client):
"""Internal function"""
message = client._userdata.popleft()
if isinstance(message, dict):
client.publish(**message)
elif isinstance(message, (tuple, list)):
client.publish(*message)
else:
raise TypeError('message must be a dict, tuple, or list')
def _on_connect(client: paho.Client, userdata: MessagesList, flags, reason_code, properties):
"""Internal v5 callback"""
if reason_code == 0:
if len(userdata) > 0:
_do_publish(client)
else:
raise mqtt.MQTTException(paho.connack_string(reason_code))
def _on_publish(
client: paho.Client, userdata: collections.deque[MessagesList], mid: int, reason_codes: ReasonCode, properties: Properties,
) -> None:
"""Internal callback"""
#pylint: disable=unused-argument
if len(userdata) == 0:
client.disconnect()
else:
_do_publish(client)
def multiple(
msgs: MessagesList,
hostname: str = "localhost",
port: int = 1883,
client_id: str = "",
keepalive: int = 60,
will: MessageDict | None = None,
auth: AuthParameter | None = None,
tls: TLSParameter | None = None,
protocol: MQTTProtocolVersion = paho.MQTTv311,
transport: Literal["tcp", "websockets"] = "tcp",
proxy_args: Any | None = None,
) -> None:
"""Publish multiple messages to a broker, then disconnect cleanly.
This function creates an MQTT client, connects to a broker and publishes a
list of messages. Once the messages have been delivered, it disconnects
cleanly from the broker.
:param msgs: a list of messages to publish. Each message is either a dict or a
tuple.
If a dict, only the topic must be present. Default values will be
used for any missing arguments. The dict must be of the form:
msg = {'topic':"<topic>", 'payload':"<payload>", 'qos':<qos>,
'retain':<retain>}
topic must be present and may not be empty.
If payload is "", None or not present then a zero length payload
will be published.
If qos is not present, the default of 0 is used.
If retain is not present, the default of False is used.
If a tuple, then it must be of the form:
("<topic>", "<payload>", qos, retain)
:param str hostname: the address of the broker to connect to.
Defaults to localhost.
:param int port: the port to connect to the broker on. Defaults to 1883.
:param str client_id: the MQTT client id to use. If "" or None, the Paho library will
generate a client id automatically.
:param int keepalive: the keepalive timeout value for the client. Defaults to 60
seconds.
:param will: a dict containing will parameters for the client: will = {'topic':
"<topic>", 'payload':"<payload">, 'qos':<qos>, 'retain':<retain>}.
Topic is required, all other parameters are optional and will
default to None, 0 and False respectively.
Defaults to None, which indicates no will should be used.
:param auth: a dict containing authentication parameters for the client:
auth = {'username':"<username>", 'password':"<password>"}
Username is required, password is optional and will default to None
if not provided.
Defaults to None, which indicates no authentication is to be used.
:param tls: a dict containing TLS configuration parameters for the client:
dict = {'ca_certs':"<ca_certs>", 'certfile':"<certfile>",
'keyfile':"<keyfile>", 'tls_version':"<tls_version>",
'ciphers':"<ciphers">, 'insecure':"<bool>"}
ca_certs is required, all other parameters are optional and will
default to None if not provided, which results in the client using
the default behaviour - see the paho.mqtt.client documentation.
Alternatively, tls input can be an SSLContext object, which will be
processed using the tls_set_context method.
Defaults to None, which indicates that TLS should not be used.
:param str transport: set to "tcp" to use the default setting of transport which is
raw TCP. Set to "websockets" to use WebSockets as the transport.
:param proxy_args: a dictionary that will be given to the client.
"""
if not isinstance(msgs, Iterable):
raise TypeError('msgs must be an iterable')
if len(msgs) == 0:
raise ValueError('msgs is empty')
client = paho.Client(
CallbackAPIVersion.VERSION2,
client_id=client_id,
userdata=collections.deque(msgs),
protocol=protocol,
transport=transport,
)
client.enable_logger()
client.on_publish = _on_publish
client.on_connect = _on_connect # type: ignore
if proxy_args is not None:
client.proxy_set(**proxy_args)
if auth:
username = auth.get('username')
if username:
password = auth.get('password')
client.username_pw_set(username, password)
else:
raise KeyError("The 'username' key was not found, this is "
"required for auth")
if will is not None:
client.will_set(**will)
if tls is not None:
if isinstance(tls, dict):
insecure = tls.pop('insecure', False)
# mypy don't get that tls no longer contains the key insecure
client.tls_set(**tls) # type: ignore[misc]
if insecure:
# Must be set *after* the `client.tls_set()` call since it sets
# up the SSL context that `client.tls_insecure_set` alters.
client.tls_insecure_set(insecure)
else:
# Assume input is SSLContext object
client.tls_set_context(tls)
client.connect(hostname, port, keepalive)
client.loop_forever()
def single(
topic: str,
payload: paho.PayloadType = None,
qos: int = 0,
retain: bool = False,
hostname: str = "localhost",
port: int = 1883,
client_id: str = "",
keepalive: int = 60,
will: MessageDict | None = None,
auth: AuthParameter | None = None,
tls: TLSParameter | None = None,
protocol: MQTTProtocolVersion = paho.MQTTv311,
transport: Literal["tcp", "websockets"] = "tcp",
proxy_args: Any | None = None,
) -> None:
"""Publish a single message to a broker, then disconnect cleanly.
This function creates an MQTT client, connects to a broker and publishes a
single message. Once the message has been delivered, it disconnects cleanly
from the broker.
:param str topic: the only required argument must be the topic string to which the
payload will be published.
:param payload: the payload to be published. If "" or None, a zero length payload
will be published.
:param int qos: the qos to use when publishing, default to 0.
:param bool retain: set the message to be retained (True) or not (False).
:param str hostname: the address of the broker to connect to.
Defaults to localhost.
:param int port: the port to connect to the broker on. Defaults to 1883.
:param str client_id: the MQTT client id to use. If "" or None, the Paho library will
generate a client id automatically.
:param int keepalive: the keepalive timeout value for the client. Defaults to 60
seconds.
:param will: a dict containing will parameters for the client: will = {'topic':
"<topic>", 'payload':"<payload">, 'qos':<qos>, 'retain':<retain>}.
Topic is required, all other parameters are optional and will
default to None, 0 and False respectively.
Defaults to None, which indicates no will should be used.
:param auth: a dict containing authentication parameters for the client:
Username is required, password is optional and will default to None
auth = {'username':"<username>", 'password':"<password>"}
if not provided.
Defaults to None, which indicates no authentication is to be used.
:param tls: a dict containing TLS configuration parameters for the client:
dict = {'ca_certs':"<ca_certs>", 'certfile':"<certfile>",
'keyfile':"<keyfile>", 'tls_version':"<tls_version>",
'ciphers':"<ciphers">, 'insecure':"<bool>"}
ca_certs is required, all other parameters are optional and will
default to None if not provided, which results in the client using
the default behaviour - see the paho.mqtt.client documentation.
Defaults to None, which indicates that TLS should not be used.
Alternatively, tls input can be an SSLContext object, which will be
processed using the tls_set_context method.
:param transport: set to "tcp" to use the default setting of transport which is
raw TCP. Set to "websockets" to use WebSockets as the transport.
:param proxy_args: a dictionary that will be given to the client.
"""
msg: MessageDict = {'topic':topic, 'payload':payload, 'qos':qos, 'retain':retain}
multiple([msg], hostname, port, client_id, keepalive, will, auth, tls,
protocol, transport, proxy_args)

0
sbapp/mqtt/py.typed Normal file
View File

223
sbapp/mqtt/reasoncodes.py Normal file
View File

@ -0,0 +1,223 @@
# *******************************************************************
# Copyright (c) 2017, 2019 IBM Corp.
#
# All rights reserved. This program and the accompanying materials
# are made available under the terms of the Eclipse Public License v2.0
# and Eclipse Distribution License v1.0 which accompany this distribution.
#
# The Eclipse Public License is available at
# http://www.eclipse.org/legal/epl-v20.html
# and the Eclipse Distribution License is available at
# http://www.eclipse.org/org/documents/edl-v10.php.
#
# Contributors:
# Ian Craggs - initial implementation and/or documentation
# *******************************************************************
import functools
import warnings
from typing import Any
from .packettypes import PacketTypes
@functools.total_ordering
class ReasonCode:
"""MQTT version 5.0 reason codes class.
See ReasonCode.names for a list of possible numeric values along with their
names and the packets to which they apply.
"""
def __init__(self, packetType: int, aName: str ="Success", identifier: int =-1):
"""
packetType: the type of the packet, such as PacketTypes.CONNECT that
this reason code will be used with. Some reason codes have different
names for the same identifier when used a different packet type.
aName: the String name of the reason code to be created. Ignored
if the identifier is set.
identifier: an integer value of the reason code to be created.
"""
self.packetType = packetType
self.names = {
0: {"Success": [PacketTypes.CONNACK, PacketTypes.PUBACK,
PacketTypes.PUBREC, PacketTypes.PUBREL, PacketTypes.PUBCOMP,
PacketTypes.UNSUBACK, PacketTypes.AUTH],
"Normal disconnection": [PacketTypes.DISCONNECT],
"Granted QoS 0": [PacketTypes.SUBACK]},
1: {"Granted QoS 1": [PacketTypes.SUBACK]},
2: {"Granted QoS 2": [PacketTypes.SUBACK]},
4: {"Disconnect with will message": [PacketTypes.DISCONNECT]},
16: {"No matching subscribers":
[PacketTypes.PUBACK, PacketTypes.PUBREC]},
17: {"No subscription found": [PacketTypes.UNSUBACK]},
24: {"Continue authentication": [PacketTypes.AUTH]},
25: {"Re-authenticate": [PacketTypes.AUTH]},
128: {"Unspecified error": [PacketTypes.CONNACK, PacketTypes.PUBACK,
PacketTypes.PUBREC, PacketTypes.SUBACK, PacketTypes.UNSUBACK,
PacketTypes.DISCONNECT], },
129: {"Malformed packet":
[PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
130: {"Protocol error":
[PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
131: {"Implementation specific error": [PacketTypes.CONNACK,
PacketTypes.PUBACK, PacketTypes.PUBREC, PacketTypes.SUBACK,
PacketTypes.UNSUBACK, PacketTypes.DISCONNECT], },
132: {"Unsupported protocol version": [PacketTypes.CONNACK]},
133: {"Client identifier not valid": [PacketTypes.CONNACK]},
134: {"Bad user name or password": [PacketTypes.CONNACK]},
135: {"Not authorized": [PacketTypes.CONNACK, PacketTypes.PUBACK,
PacketTypes.PUBREC, PacketTypes.SUBACK, PacketTypes.UNSUBACK,
PacketTypes.DISCONNECT], },
136: {"Server unavailable": [PacketTypes.CONNACK]},
137: {"Server busy": [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
138: {"Banned": [PacketTypes.CONNACK]},
139: {"Server shutting down": [PacketTypes.DISCONNECT]},
140: {"Bad authentication method":
[PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
141: {"Keep alive timeout": [PacketTypes.DISCONNECT]},
142: {"Session taken over": [PacketTypes.DISCONNECT]},
143: {"Topic filter invalid":
[PacketTypes.SUBACK, PacketTypes.UNSUBACK, PacketTypes.DISCONNECT]},
144: {"Topic name invalid":
[PacketTypes.CONNACK, PacketTypes.PUBACK,
PacketTypes.PUBREC, PacketTypes.DISCONNECT]},
145: {"Packet identifier in use":
[PacketTypes.PUBACK, PacketTypes.PUBREC,
PacketTypes.SUBACK, PacketTypes.UNSUBACK]},
146: {"Packet identifier not found":
[PacketTypes.PUBREL, PacketTypes.PUBCOMP]},
147: {"Receive maximum exceeded": [PacketTypes.DISCONNECT]},
148: {"Topic alias invalid": [PacketTypes.DISCONNECT]},
149: {"Packet too large": [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
150: {"Message rate too high": [PacketTypes.DISCONNECT]},
151: {"Quota exceeded": [PacketTypes.CONNACK, PacketTypes.PUBACK,
PacketTypes.PUBREC, PacketTypes.SUBACK, PacketTypes.DISCONNECT], },
152: {"Administrative action": [PacketTypes.DISCONNECT]},
153: {"Payload format invalid":
[PacketTypes.PUBACK, PacketTypes.PUBREC, PacketTypes.DISCONNECT]},
154: {"Retain not supported":
[PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
155: {"QoS not supported":
[PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
156: {"Use another server":
[PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
157: {"Server moved":
[PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
158: {"Shared subscription not supported":
[PacketTypes.SUBACK, PacketTypes.DISCONNECT]},
159: {"Connection rate exceeded":
[PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
160: {"Maximum connect time":
[PacketTypes.DISCONNECT]},
161: {"Subscription identifiers not supported":
[PacketTypes.SUBACK, PacketTypes.DISCONNECT]},
162: {"Wildcard subscription not supported":
[PacketTypes.SUBACK, PacketTypes.DISCONNECT]},
}
if identifier == -1:
if packetType == PacketTypes.DISCONNECT and aName == "Success":
aName = "Normal disconnection"
self.set(aName)
else:
self.value = identifier
self.getName() # check it's good
def __getName__(self, packetType, identifier):
"""
Get the reason code string name for a specific identifier.
The name can vary by packet type for the same identifier, which
is why the packet type is also required.
Used when displaying the reason code.
"""
if identifier not in self.names:
raise KeyError(identifier)
names = self.names[identifier]
namelist = [name for name in names.keys() if packetType in names[name]]
if len(namelist) != 1:
raise ValueError(f"Expected exactly one name, found {namelist!r}")
return namelist[0]
def getId(self, name):
"""
Get the numeric id corresponding to a reason code name.
Used when setting the reason code for a packetType
check that only valid codes for the packet are set.
"""
for code in self.names.keys():
if name in self.names[code].keys():
if self.packetType in self.names[code][name]:
return code
raise KeyError(f"Reason code name not found: {name}")
def set(self, name):
self.value = self.getId(name)
def unpack(self, buffer):
c = buffer[0]
name = self.__getName__(self.packetType, c)
self.value = self.getId(name)
return 1
def getName(self):
"""Returns the reason code name corresponding to the numeric value which is set.
"""
return self.__getName__(self.packetType, self.value)
def __eq__(self, other):
if isinstance(other, int):
return self.value == other
if isinstance(other, str):
return other == str(self)
if isinstance(other, ReasonCode):
return self.value == other.value
return False
def __lt__(self, other):
if isinstance(other, int):
return self.value < other
if isinstance(other, ReasonCode):
return self.value < other.value
return NotImplemented
def __repr__(self):
try:
packet_name = PacketTypes.Names[self.packetType]
except IndexError:
packet_name = "Unknown"
return f"ReasonCode({packet_name}, {self.getName()!r})"
def __str__(self):
return self.getName()
def json(self):
return self.getName()
def pack(self):
return bytearray([self.value])
@property
def is_failure(self) -> bool:
return self.value >= 0x80
class _CompatibilityIsInstance(type):
def __instancecheck__(self, other: Any) -> bool:
return isinstance(other, ReasonCode)
class ReasonCodes(ReasonCode, metaclass=_CompatibilityIsInstance):
def __init__(self, *args, **kwargs):
warnings.warn("ReasonCodes is deprecated, use ReasonCode (singular) instead",
category=DeprecationWarning,
stacklevel=2,
)
super().__init__(*args, **kwargs)

281
sbapp/mqtt/subscribe.py Normal file
View File

@ -0,0 +1,281 @@
# Copyright (c) 2016 Roger Light <roger@atchoo.org>
#
# All rights reserved. This program and the accompanying materials
# are made available under the terms of the Eclipse Public License v2.0
# and Eclipse Distribution License v1.0 which accompany this distribution.
#
# The Eclipse Public License is available at
# http://www.eclipse.org/legal/epl-v20.html
# and the Eclipse Distribution License is available at
# http://www.eclipse.org/org/documents/edl-v10.php.
#
# Contributors:
# Roger Light - initial API and implementation
"""
This module provides some helper functions to allow straightforward subscribing
to topics and retrieving messages. The two functions are simple(), which
returns one or messages matching a set of topics, and callback() which allows
you to pass a callback for processing of messages.
"""
from .. import mqtt
from . import client as paho
def _on_connect(client, userdata, flags, reason_code, properties):
"""Internal callback"""
if reason_code != 0:
raise mqtt.MQTTException(paho.connack_string(reason_code))
if isinstance(userdata['topics'], list):
for topic in userdata['topics']:
client.subscribe(topic, userdata['qos'])
else:
client.subscribe(userdata['topics'], userdata['qos'])
def _on_message_callback(client, userdata, message):
"""Internal callback"""
userdata['callback'](client, userdata['userdata'], message)
def _on_message_simple(client, userdata, message):
"""Internal callback"""
if userdata['msg_count'] == 0:
return
# Don't process stale retained messages if 'retained' was false
if message.retain and not userdata['retained']:
return
userdata['msg_count'] = userdata['msg_count'] - 1
if userdata['messages'] is None and userdata['msg_count'] == 0:
userdata['messages'] = message
client.disconnect()
return
userdata['messages'].append(message)
if userdata['msg_count'] == 0:
client.disconnect()
def callback(callback, topics, qos=0, userdata=None, hostname="localhost",
port=1883, client_id="", keepalive=60, will=None, auth=None,
tls=None, protocol=paho.MQTTv311, transport="tcp",
clean_session=True, proxy_args=None):
"""Subscribe to a list of topics and process them in a callback function.
This function creates an MQTT client, connects to a broker and subscribes
to a list of topics. Incoming messages are processed by the user provided
callback. This is a blocking function and will never return.
:param callback: function with the same signature as `on_message` for
processing the messages received.
:param topics: either a string containing a single topic to subscribe to, or a
list of topics to subscribe to.
:param int qos: the qos to use when subscribing. This is applied to all topics.
:param userdata: passed to the callback
:param str hostname: the address of the broker to connect to.
Defaults to localhost.
:param int port: the port to connect to the broker on. Defaults to 1883.
:param str client_id: the MQTT client id to use. If "" or None, the Paho library will
generate a client id automatically.
:param int keepalive: the keepalive timeout value for the client. Defaults to 60
seconds.
:param will: a dict containing will parameters for the client: will = {'topic':
"<topic>", 'payload':"<payload">, 'qos':<qos>, 'retain':<retain>}.
Topic is required, all other parameters are optional and will
default to None, 0 and False respectively.
Defaults to None, which indicates no will should be used.
:param auth: a dict containing authentication parameters for the client:
auth = {'username':"<username>", 'password':"<password>"}
Username is required, password is optional and will default to None
if not provided.
Defaults to None, which indicates no authentication is to be used.
:param tls: a dict containing TLS configuration parameters for the client:
dict = {'ca_certs':"<ca_certs>", 'certfile':"<certfile>",
'keyfile':"<keyfile>", 'tls_version':"<tls_version>",
'ciphers':"<ciphers">, 'insecure':"<bool>"}
ca_certs is required, all other parameters are optional and will
default to None if not provided, which results in the client using
the default behaviour - see the paho.mqtt.client documentation.
Alternatively, tls input can be an SSLContext object, which will be
processed using the tls_set_context method.
Defaults to None, which indicates that TLS should not be used.
:param str transport: set to "tcp" to use the default setting of transport which is
raw TCP. Set to "websockets" to use WebSockets as the transport.
:param clean_session: a boolean that determines the client type. If True,
the broker will remove all information about this client
when it disconnects. If False, the client is a persistent
client and subscription information and queued messages
will be retained when the client disconnects.
Defaults to True.
:param proxy_args: a dictionary that will be given to the client.
"""
if qos < 0 or qos > 2:
raise ValueError('qos must be in the range 0-2')
callback_userdata = {
'callback':callback,
'topics':topics,
'qos':qos,
'userdata':userdata}
client = paho.Client(
paho.CallbackAPIVersion.VERSION2,
client_id=client_id,
userdata=callback_userdata,
protocol=protocol,
transport=transport,
clean_session=clean_session,
)
client.enable_logger()
client.on_message = _on_message_callback
client.on_connect = _on_connect
if proxy_args is not None:
client.proxy_set(**proxy_args)
if auth:
username = auth.get('username')
if username:
password = auth.get('password')
client.username_pw_set(username, password)
else:
raise KeyError("The 'username' key was not found, this is "
"required for auth")
if will is not None:
client.will_set(**will)
if tls is not None:
if isinstance(tls, dict):
insecure = tls.pop('insecure', False)
client.tls_set(**tls)
if insecure:
# Must be set *after* the `client.tls_set()` call since it sets
# up the SSL context that `client.tls_insecure_set` alters.
client.tls_insecure_set(insecure)
else:
# Assume input is SSLContext object
client.tls_set_context(tls)
client.connect(hostname, port, keepalive)
client.loop_forever()
def simple(topics, qos=0, msg_count=1, retained=True, hostname="localhost",
port=1883, client_id="", keepalive=60, will=None, auth=None,
tls=None, protocol=paho.MQTTv311, transport="tcp",
clean_session=True, proxy_args=None):
"""Subscribe to a list of topics and return msg_count messages.
This function creates an MQTT client, connects to a broker and subscribes
to a list of topics. Once "msg_count" messages have been received, it
disconnects cleanly from the broker and returns the messages.
:param topics: either a string containing a single topic to subscribe to, or a
list of topics to subscribe to.
:param int qos: the qos to use when subscribing. This is applied to all topics.
:param int msg_count: the number of messages to retrieve from the broker.
if msg_count == 1 then a single MQTTMessage will be returned.
if msg_count > 1 then a list of MQTTMessages will be returned.
:param bool retained: If set to True, retained messages will be processed the same as
non-retained messages. If set to False, retained messages will
be ignored. This means that with retained=False and msg_count=1,
the function will return the first message received that does
not have the retained flag set.
:param str hostname: the address of the broker to connect to.
Defaults to localhost.
:param int port: the port to connect to the broker on. Defaults to 1883.
:param str client_id: the MQTT client id to use. If "" or None, the Paho library will
generate a client id automatically.
:param int keepalive: the keepalive timeout value for the client. Defaults to 60
seconds.
:param will: a dict containing will parameters for the client: will = {'topic':
"<topic>", 'payload':"<payload">, 'qos':<qos>, 'retain':<retain>}.
Topic is required, all other parameters are optional and will
default to None, 0 and False respectively.
Defaults to None, which indicates no will should be used.
:param auth: a dict containing authentication parameters for the client:
auth = {'username':"<username>", 'password':"<password>"}
Username is required, password is optional and will default to None
if not provided.
Defaults to None, which indicates no authentication is to be used.
:param tls: a dict containing TLS configuration parameters for the client:
dict = {'ca_certs':"<ca_certs>", 'certfile':"<certfile>",
'keyfile':"<keyfile>", 'tls_version':"<tls_version>",
'ciphers':"<ciphers">, 'insecure':"<bool>"}
ca_certs is required, all other parameters are optional and will
default to None if not provided, which results in the client using
the default behaviour - see the paho.mqtt.client documentation.
Alternatively, tls input can be an SSLContext object, which will be
processed using the tls_set_context method.
Defaults to None, which indicates that TLS should not be used.
:param protocol: the MQTT protocol version to use. Defaults to MQTTv311.
:param transport: set to "tcp" to use the default setting of transport which is
raw TCP. Set to "websockets" to use WebSockets as the transport.
:param clean_session: a boolean that determines the client type. If True,
the broker will remove all information about this client
when it disconnects. If False, the client is a persistent
client and subscription information and queued messages
will be retained when the client disconnects.
Defaults to True. If protocol is MQTTv50, clean_session
is ignored.
:param proxy_args: a dictionary that will be given to the client.
"""
if msg_count < 1:
raise ValueError('msg_count must be > 0')
# Set ourselves up to return a single message if msg_count == 1, or a list
# if > 1.
if msg_count == 1:
messages = None
else:
messages = []
# Ignore clean_session if protocol is MQTTv50, otherwise Client will raise
if protocol == paho.MQTTv5:
clean_session = None
userdata = {'retained':retained, 'msg_count':msg_count, 'messages':messages}
callback(_on_message_simple, topics, qos, userdata, hostname, port,
client_id, keepalive, will, auth, tls, protocol, transport,
clean_session, proxy_args)
return userdata['messages']

113
sbapp/mqtt/subscribeoptions.py Normal file
View File

@ -0,0 +1,113 @@
"""
*******************************************************************
Copyright (c) 2017, 2019 IBM Corp.
All rights reserved. This program and the accompanying materials
are made available under the terms of the Eclipse Public License v2.0
and Eclipse Distribution License v1.0 which accompany this distribution.
The Eclipse Public License is available at
http://www.eclipse.org/legal/epl-v20.html
and the Eclipse Distribution License is available at
http://www.eclipse.org/org/documents/edl-v10.php.
Contributors:
Ian Craggs - initial implementation and/or documentation
*******************************************************************
"""
class MQTTException(Exception):
pass
class SubscribeOptions:
"""The MQTT v5.0 subscribe options class.
The options are:
qos: As in MQTT v3.1.1.
noLocal: True or False. If set to True, the subscriber will not receive its own publications.
retainAsPublished: True or False. If set to True, the retain flag on received publications will be as set
by the publisher.
retainHandling: RETAIN_SEND_ON_SUBSCRIBE, RETAIN_SEND_IF_NEW_SUB or RETAIN_DO_NOT_SEND
Controls when the broker should send retained messages:
- RETAIN_SEND_ON_SUBSCRIBE: on any successful subscribe request
- RETAIN_SEND_IF_NEW_SUB: only if the subscribe request is new
- RETAIN_DO_NOT_SEND: never send retained messages
"""
# retain handling options
RETAIN_SEND_ON_SUBSCRIBE, RETAIN_SEND_IF_NEW_SUB, RETAIN_DO_NOT_SEND = range(
0, 3)
def __init__(
self,
qos: int = 0,
noLocal: bool = False,
retainAsPublished: bool = False,
retainHandling: int = RETAIN_SEND_ON_SUBSCRIBE,
):
"""
qos: 0, 1 or 2. 0 is the default.
noLocal: True or False. False is the default and corresponds to MQTT v3.1.1 behavior.
retainAsPublished: True or False. False is the default and corresponds to MQTT v3.1.1 behavior.
retainHandling: RETAIN_SEND_ON_SUBSCRIBE, RETAIN_SEND_IF_NEW_SUB or RETAIN_DO_NOT_SEND
RETAIN_SEND_ON_SUBSCRIBE is the default and corresponds to MQTT v3.1.1 behavior.
"""
object.__setattr__(self, "names",
["QoS", "noLocal", "retainAsPublished", "retainHandling"])
self.QoS = qos # bits 0,1
self.noLocal = noLocal # bit 2
self.retainAsPublished = retainAsPublished # bit 3
self.retainHandling = retainHandling # bits 4 and 5: 0, 1 or 2
if self.retainHandling not in (0, 1, 2):
raise AssertionError(f"Retain handling should be 0, 1 or 2, not {self.retainHandling}")
if self.QoS not in (0, 1, 2):
raise AssertionError(f"QoS should be 0, 1 or 2, not {self.QoS}")
def __setattr__(self, name, value):
if name not in self.names:
raise MQTTException(
f"{name} Attribute name must be one of {self.names}")
object.__setattr__(self, name, value)
def pack(self):
if self.retainHandling not in (0, 1, 2):
raise AssertionError(f"Retain handling should be 0, 1 or 2, not {self.retainHandling}")
if self.QoS not in (0, 1, 2):
raise AssertionError(f"QoS should be 0, 1 or 2, not {self.QoS}")
noLocal = 1 if self.noLocal else 0
retainAsPublished = 1 if self.retainAsPublished else 0
data = [(self.retainHandling << 4) | (retainAsPublished << 3) |
(noLocal << 2) | self.QoS]
return bytes(data)
def unpack(self, buffer):
b0 = buffer[0]
self.retainHandling = ((b0 >> 4) & 0x03)
self.retainAsPublished = True if ((b0 >> 3) & 0x01) == 1 else False
self.noLocal = True if ((b0 >> 2) & 0x01) == 1 else False
self.QoS = (b0 & 0x03)
if self.retainHandling not in (0, 1, 2):
raise AssertionError(f"Retain handling should be 0, 1 or 2, not {self.retainHandling}")
if self.QoS not in (0, 1, 2):
raise AssertionError(f"QoS should be 0, 1 or 2, not {self.QoS}")
return 1
def __repr__(self):
return str(self)
def __str__(self):
return "{QoS="+str(self.QoS)+", noLocal="+str(self.noLocal) +\
", retainAsPublished="+str(self.retainAsPublished) +\
", retainHandling="+str(self.retainHandling)+"}"
def json(self):
data = {
"QoS": self.QoS,
"noLocal": self.noLocal,
"retainAsPublished": self.retainAsPublished,
"retainHandling": self.retainHandling,
}
return data