1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
|
# SPDX-License-Identifier: BSD-3-Clause
from __future__ import annotations
import dataclasses
import sys
if sys.version_info >= (3, 10):
from typing import TypeAlias
else:
from typing_extensions import TypeAlias
MAX_TOPIC_LENGTH = 65535
@dataclasses.dataclass(frozen=True)
class Wildcard:
"""MQTT wildcard that can be subscribed to, but not published to.
A wildcard is similar to a topic, but can optionally contain ``+`` and ``#``
placeholders. You can access the ``value`` attribute directly to perform ``str``
operations on a wildcard.
Args:
value: The wildcard string.
Attributes:
value: The wildcard string.
"""
value: str
def __str__(self) -> str:
return self.value
def __post_init__(self) -> None:
"""Validate the wildcard."""
if not isinstance(self.value, str):
msg = "Wildcard must be of type str"
raise TypeError(msg)
if (
len(self.value) == 0
or len(self.value) > MAX_TOPIC_LENGTH
or "#/" in self.value
or any(
"+" in level or "#" in level
for level in self.value.split("/")
if len(level) > 1
)
):
msg = f"Invalid wildcard: {self.value}"
raise ValueError(msg)
WildcardLike: TypeAlias = "str | Wildcard"
@dataclasses.dataclass(frozen=True)
class Topic(Wildcard):
"""MQTT topic that can be published and subscribed to.
Args:
value: The topic string.
Attributes:
value: The topic string.
"""
def __post_init__(self) -> None:
"""Validate the topic."""
if not isinstance(self.value, str):
msg = "Topic must be of type str"
raise TypeError(msg)
if (
len(self.value) == 0
or len(self.value) > MAX_TOPIC_LENGTH
or "+" in self.value
or "#" in self.value
):
msg = f"Invalid topic: {self.value}"
raise ValueError(msg)
def matches(self, wildcard: WildcardLike) -> bool:
"""Check if the topic matches a given wildcard.
Args:
wildcard: The wildcard to match against.
Returns:
True if the topic matches the wildcard, False otherwise.
"""
if not isinstance(wildcard, Wildcard):
wildcard = Wildcard(wildcard)
# Split topics into levels to compare them one by one
topic_levels = self.value.split("/")
wildcard_levels = str(wildcard).split("/")
if wildcard_levels[0] == "$share":
# Shared subscriptions use the topic structure: $share/<group_id>/<topic>
wildcard_levels = wildcard_levels[2:]
def recurse(tl: list[str], wl: list[str]) -> bool:
"""Recursively match topic levels with wildcard levels."""
if not tl:
if not wl or wl[0] == "#":
return True
return False
if not wl:
return False
if wl[0] == "#":
return True
if tl[0] == wl[0] or wl[0] == "+":
return recurse(tl[1:], wl[1:])
return False
return recurse(topic_levels, wildcard_levels)
TopicLike: TypeAlias = "str | Topic"
|
