feat: give the rooms a line to the church broker
The first piece of turning these machines into connected endpoints. classroom-mqtt is one small service per room that subscribes, validates what arrives, and hands the work to commands that already exist. Nothing in the room depends on it. A broker that is down, a network that is missing, a machine with no broker configured at all — each is a service sitting quietly, not a classroom that fails. Nothing waits for it at boot and the shipped default is off, so every existing classroom upgrades into this release with messaging disabled until a broker exists. The room is the hostname, which is already how these machines are named. Topics carry no product name because the broker is the church's and these are just rooms: classroom/notify/all, classroom/notify/<room>, and a retained classroom/status/<room> so anyone asking later can be told which rooms are up without waiting for them to say so again. Notifications are events and are never retained; status is state and is. Stale messages cannot come back, which is the property that matters most and has two halves. The session is clean, so a broker never hands over a backlog from while the room was off. And retained notifications are refused on arrival, because otherwise one would replay into every room at every reboot forever. The honest limit is written down: MQTT only flags a retained message when it is replayed to a new subscription, so a retained publish delivered live is indistinguishable from an ordinary one and is shown once — which is right, since it is live at that moment, and the replay is the actual hazard. Delivery reuses the session-aware sender rather than a second copy of it. classroom-notify-test grew into the real thing, so it is now classroom-notify with the sample sender a wrapper over it, and its 36 tests came along. The broker password lives in device.conf, so that file stops being world readable: 0640 root:classroom, which everything that reads it already satisfies. Transport logic is kept free of any broker library so it can be tested without one — 40 checks on room identity, the topic contract and which payloads become notifications. Separately verified against a real mosquitto: connect, subscribe, retained status, per-room and all-room delivery, urgency, duplicate and expired and malformed messages ignored, the broker killed with the daemon surviving, reconnect, and a stale retained notification left on the broker not reaching the screen when a room boots. Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
ekklesia
co-authored by
Claude Opus 5
Claude Opus 5
parent
c3106edd30
commit
dfd42f6fdf
@@ -0,0 +1,177 @@
|
||||
"""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/<room> one room
|
||||
classroom/status/<room> 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",
|
||||
)
|
||||
Reference in New Issue
Block a user