"""The classroom's side of the church message broker. Room identity, the topic contract, and the rules for what counts as an acceptable message. Deliberately free of any broker library, so the decisions can be tested without one running. Topics carry no product name. The broker belongs to the church and these are just rooms: classroom/notify/all every room classroom/notify/ one room classroom/status/ retained; whether that room is up Notifications are events, never retained. Status is state, and is. """ import collections import json import re import socket import time NOTIFY_ALL = "classroom/notify/all" NOTIFY_ROOM = "classroom/notify/{room}" STATUS_ROOM = "classroom/status/{room}" URGENCIES = ("low", "normal", "critical") DEFAULT_URGENCY = "normal" DEFAULT_PORT = 1883 # Room names become topic segments, so they have to be plain. UNSAFE = re.compile(r"[^a-z0-9_-]+") def room_name(config=None, hostname=None): """Which room this machine is. The hostname, because that is already how these machines are named — nursery, k2, older. ROOM in device.conf overrides it for the odd machine whose hostname cannot be changed. """ config = config or {} room = (config.get("ROOM") or "").strip() if not room: room = socket.gethostname() if hostname is None else hostname room = (room or "").split(".")[0].strip().lower() room = UNSAFE.sub("-", room).strip("-") return room or "unknown" def notify_topics(room): """Everything a room listens to for notifications.""" return (NOTIFY_ALL, NOTIFY_ROOM.format(room=room)) def status_topic(room): return STATUS_ROOM.format(room=room) def broker(config): """Broker settings, or None when this classroom has no broker yet.""" config = config or {} if (config.get("MQTT_ENABLED") or "").strip().lower() not in ( "1", "true", "yes", "on", "enabled", ): return None host = (config.get("MQTT_HOST") or "").strip() if not host: return None try: port = int((config.get("MQTT_PORT") or DEFAULT_PORT)) except (TypeError, ValueError): port = DEFAULT_PORT return { "host": host, "port": port, "username": (config.get("MQTT_USERNAME") or "").strip() or None, "password": config.get("MQTT_PASSWORD") or None, } class Notifications: """Decides which incoming payloads become notifications. Every rejection is ordinary traffic rather than an error: a malformed, stale or repeated message is something to drop and count, never something to fall over on. """ def __init__(self, remember=512): self.remember = remember self.seen = collections.OrderedDict() def accept(self, payload, retained=False, now=None): """Returns (notification, reason). notification is None if dropped.""" now = time.time() if now is None else now # A retained notification is a stale one by definition: it would # arrive again at every reconnect and every reboot. if retained: return None, "retained" if isinstance(payload, (bytes, bytearray)): try: payload = payload.decode("utf-8") except UnicodeDecodeError: return None, "not utf-8" try: data = json.loads(payload) except (TypeError, ValueError): return None, "not json" if not isinstance(data, dict): return None, "not an object" title = data.get("title") if not isinstance(title, str) or not title.strip(): return None, "no title" message = data.get("message", "") if not isinstance(message, str): return None, "message is not text" expires = data.get("expires_at") if expires is not None: if isinstance(expires, bool) or not isinstance(expires, (int, float)): return None, "expires_at is not a time" if expires <= now: return None, "expired" identifier = data.get("id") if identifier is not None: if not isinstance(identifier, str): return None, "id is not text" if identifier in self.seen: return None, "duplicate" self.seen[identifier] = now while len(self.seen) > self.remember: self.seen.popitem(last=False) urgency = data.get("urgency", DEFAULT_URGENCY) if urgency not in URGENCIES: urgency = DEFAULT_URGENCY return ( { "id": identifier, "title": title.strip(), "message": message.strip(), "urgency": urgency, }, "ok", )