Use an existing Mosquitto 2.1.2 executable and a disposable Python environment with paho-mqtt==2.1.0. Do not install a system service for this lab. The archive contains the complete program, the test suite, broker configuration and blank worksheet; the program below is identical to its download. The Paho dependency is pinned but is not bundled into the experiment archive.
The first test should normally show broker acknowledgement and UNKNOWN. The two-terminal run can produce RECEIPT. A missing executable, occupied port, refused connection or failed subscription is an error rather than a fabricated successful result. Every program connection is closed in a finally block. There is no automatic reconnect or automatic resend after failure; decide what an unknown outcome means before retrying a real action.
Complete door_ack_lab.py
"""Finite loopback MQTT application-receipt experiment. No GPIO/actuation.
CPython 3.11+, paho-mqtt==2.1.0; Mosquitto 2.1.2 is a separate process.
Only 127.0.0.1 is used. Receiver cache is volatile, bounded, not durable.
"""
import argparse
import json
import math
import re
import secrets
import time
ALERT = 'tricky105/alert'
RECEIPT = 'tricky105/receipt'
HEX = re.compile(r'[0-9a-f]{32}\Z')
def finite(value, low, high):
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError('NUMBER')
if not math.isfinite(value) or not low <= value <= high:
raise ValueError('RANGE')
return value
def encode(obj):
return json.dumps(obj, sort_keys=True, separators=(',', ':')).encode('ascii')
def decode(raw, kind):
if not isinstance(raw, bytes) or not 1 <= len(raw) <= 256:
raise ValueError('SIZE')
def pairs(items):
d = {}
for k, v in items:
if k in d:
raise ValueError('DUPLICATE_KEY')
d[k] = v
return d
d = json.loads(raw.decode('ascii'), object_pairs_hook=pairs)
keys = {'v', 'id', 'door', 'state'} if kind == 'alert' else {'v', 'id', 'receiver', 'result'}
if not isinstance(d, dict) or set(d) != keys or type(d['v']) is not int or d['v'] != 1:
raise ValueError('SCHEMA')
if not isinstance(d['id'], str) or not HEX.fullmatch(d['id']):
raise ValueError('ID')
if kind == 'alert':
if d['door'] != 'lab-door' or d['state'] not in ('OPEN', 'CLOSED'):
raise ValueError('ALERT')
elif d['receiver'] != 'room-b' or d['result'] != 'DISPLAYED':
raise ValueError('RECEIPT')
return d
class Sender:
def __init__(self, alert_id, started, timeout):
if not isinstance(alert_id, str) or not HEX.fullmatch(alert_id):
raise ValueError('ID')
finite(started, 0, 1e12)
finite(timeout, .05, 5)
self.id, self.start, self.deadline = alert_id, started, started + timeout
self.last = started
self.broker_ack = False
self.application = 'WAITING'
self.fault = None
def clock(self, now):
try:
finite(now, self.last, 1e12)
except ValueError:
self.fault, self.application = 'CLOCK', 'UNKNOWN'
return False
self.last = now
if self.application == 'WAITING' and now >= self.deadline:
self.application = 'UNKNOWN'
return not self.fault
def accept(self, raw, retained, now):
if not self.clock(now) or self.application != 'WAITING' or retained is not False:
return False
try:
d = decode(raw, 'receipt')
except (ValueError, UnicodeError):
return False
if d['id'] != self.id:
return False
self.application = 'RECEIPT'
return True
def snapshot(self, now):
self.clock(now)
return {'id': self.id, 'broker_ack': self.broker_ack,
'application': self.application, 'fault': self.fault}
class Receiver:
def __init__(self, delay=0):
self.delay = finite(delay, 0, 5)
self.cache = {} # ID -> (fingerprint, due, displayed)
self.last = 0
def receive(self, raw, retained, now):
finite(now, self.last, 1e12)
self.last = now
if retained is not False:
return None
try:
d = decode(raw, 'alert')
except (ValueError, UnicodeError):
return None
fingerprint = encode(d)
old = self.cache.get(d['id'])
if old:
if old[0] != fingerprint:
return None # ID collision/conflict: do not acknowledge
return self.receipt(d['id']) if old[2] else None
if len(self.cache) >= 32:
return None # fail closed; no eviction within this process
self.cache[d['id']] = [fingerprint, now + self.delay, False]
return None
@staticmethod
def receipt(alert_id):
return encode({'v': 1, 'id': alert_id, 'receiver': 'room-b', 'result': 'DISPLAYED'})
def due(self, now, display):
finite(now, self.last, 1e12)
self.last = now
replies = []
for alert_id, item in self.cache.items():
if not item[2] and now >= item[1]:
display(item[0].decode('ascii')) # failure => no displayed flag/receipt
item[2] = True
replies.append(self.receipt(alert_id))
return replies
def connect(port, callback, topic):
import paho.mqtt.client as mqtt
if type(port) is not int or not 1024 <= port <= 65535:
raise ValueError('PORT')
c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id='t105-' + secrets.token_hex(8),
protocol=mqtt.MQTTv311, clean_session=True, reconnect_on_failure=False)
c.connect_timeout = 2
ready, subscribed = [], []
def on_connect(client, userdata, flags, reason, properties):
if reason.is_failure:
raise RuntimeError('CONNACK')
ready.append(True)
rc, mid = client.subscribe(topic, qos=1)
if rc != mqtt.MQTT_ERR_SUCCESS:
raise RuntimeError('SUBSCRIBE')
def on_subscribe(client, userdata, mid, reasons, properties):
if not reasons or any(r.is_failure for r in reasons):
raise RuntimeError('SUBACK')
subscribed.append(True)
c.on_connect, c.on_subscribe = on_connect, on_subscribe
c.on_message = callback
try:
c.connect('127.0.0.1', port, keepalive=10)
end = time.monotonic() + 3
while not (ready and subscribed):
pump(c)
if time.monotonic() >= end:
raise TimeoutError('CONNECT_OR_SUBACK')
return c
except BaseException:
c.disconnect()
raise
def pump(c):
import paho.mqtt.client as mqtt
if c.loop(timeout=.01) != mqtt.MQTT_ERR_SUCCESS:
raise ConnectionError('LOOP_DISCONNECTED')
def publish(c, topic, payload):
info = c.publish(topic, payload, qos=1, retain=False)
if info.rc != 0:
raise ConnectionError('PUBLISH')
return info
def send(port=18885, timeout=1, state='OPEN', alert_id=None):
finite(timeout, .05, 5)
if state not in ('OPEN', 'CLOSED'):
raise ValueError('STATE')
alert_id = alert_id or secrets.token_hex(16)
holder = []
def callback(c, u, m):
if holder:
holder[0].accept(bytes(m.payload), bool(m.retain), time.monotonic())
c = connect(port, callback, RECEIPT)
try:
s = Sender(alert_id, time.monotonic(), timeout)
holder.append(s)
info = publish(c, ALERT, encode({'v': 1, 'id': alert_id, 'door': 'lab-door', 'state': state}))
while s.application == 'WAITING':
pump(c)
s.broker_ack = info.is_published()
s.clock(time.monotonic())
return s.snapshot(time.monotonic())
finally:
c.disconnect()
def receive(port=18885, duration=5, delay=0, display=None):
finite(duration, .05, 30)
r = Receiver(delay)
def show(s):
print('DISPLAY ' + s, flush=True)
display = display or show
pending = []
def callback(c, u, m):
reply = r.receive(bytes(m.payload), bool(m.retain), time.monotonic())
if reply:
pending.append(reply)
c = connect(port, callback, ALERT)
try:
print('READY room-b', flush=True)
end = time.monotonic() + duration
while time.monotonic() < end:
pump(c)
pending.extend(r.due(time.monotonic(), display))
for payload in pending:
publish(c, RECEIPT, payload) # no blocking wait inside callback
pending.clear()
return len(r.cache)
finally:
c.disconnect()
def main():
p = argparse.ArgumentParser(description=__doc__)
p.add_argument('role', choices=['sender', 'receiver'])
p.add_argument('--port', type=int, default=18885)
p.add_argument('--timeout', type=float, default=1)
p.add_argument('--duration', type=float, default=5)
p.add_argument('--delay', type=float, default=0)
p.add_argument('--id', default=None)
p.add_argument('--state', choices=['OPEN', 'CLOSED'], default='OPEN')
a = p.parse_args()
if a.role == 'sender':
print(json.dumps(send(a.port, a.timeout, a.state, a.id)), flush=True)
else:
receive(a.port, a.duration, a.delay)
if __name__ == '__main__':
main()
