"""Who owns the truth about a parent call. Pure and synchronous: every input is a method call carrying the current time, and every output is an Effect for someone else to perform. No broker, no HTTP, no clock of its own — which is what makes the awkward cases (two rooms calling at once, an approval arriving after expiry, a clear timer firing after the next call) testable properly. Deliberately in memory only. If Control loses power mid-service, the queue must die with it: a parent call that surfaces ten minutes late is worse than one that was lost. """ from __future__ import annotations from collections import OrderedDict from dataclasses import dataclass from . import contract from .validation import Request # -- effects ----------------------------------------------------------- @dataclass class PublishStatus: request_id: str room: str status: str detail: str = None @dataclass class PublishPending: request: Request waiting: int @dataclass class Deliver: """Send this to ProPresenter. serial identifies this delivery.""" request: Request serial: int @dataclass class ScheduleClear: serial: int after: float @dataclass class ClearNow: serial: int class ParentCall: def __init__(self, settings, mode=None): self.settings = settings self.mode = mode if mode in contract.MODES else settings.default_mode self.queue = [] self.serial = 0 # What each id ended up as, so a classroom that retries the same # request gets the same answer instead of a second call going out. self.outcomes = OrderedDict() self.remember = 512 # -- helpers ------------------------------------------------------- def _record(self, request_id, room, status): self.outcomes[request_id] = (room, status) self.outcomes.move_to_end(request_id) while len(self.outcomes) > self.remember: self.outcomes.popitem(last=False) def _status(self, request_id, room, status, detail=None): self._record(request_id, room, status) return PublishStatus(request_id, room, status, detail) def _pending(self): head = self.queue[0] if self.queue else None return PublishPending(head, len(self.queue)) def _start_delivery(self, request): self.serial += 1 return [ self._status(request.id, request.room, contract.SENDING), Deliver(request, self.serial), ] def _too_old(self, request, now): return (now - request.created_at) > self.settings.request_ttl # -- inputs -------------------------------------------------------- def set_mode(self, mode, now): """Modes are operator state and survive a restart, because they are retained on the broker rather than held here.""" if mode not in contract.MODES or mode == self.mode: return [] self.mode = mode # Turning Parent Call off empties the queue rather than holding # requests that no one is going to approve. if mode != contract.MODE_APPROVAL and self.queue: effects = [ self._status( held.id, held.room, contract.UNAVAILABLE, "Parent Call was switched off", ) for held in self.queue ] self.queue = [] return effects + [self._pending()] return [] def submit(self, request, now): # A repeat of an id we have already answered gets the same answer # again and is never delivered twice. Classrooms retry. if request.id in self.outcomes: room, status = self.outcomes[request.id] return [PublishStatus(request.id, room, status, "already received")] if self._too_old(request, now): return [self._status(request.id, request.room, contract.EXPIRED)] if self.mode == contract.MODE_OFF: return [self._status(request.id, request.room, contract.UNAVAILABLE)] received = self._status(request.id, request.room, contract.SUBMITTED) if self.mode == contract.MODE_AUTO: return [received] + self._start_delivery(request) self.queue.append(request) return [ received, self._status(request.id, request.room, contract.WAITING_APPROVAL), self._pending(), ] def approve(self, request_id, now): """Approve one identified request. By id, never "whatever is showing", so that two rooms calling at the same moment cannot approve each other's child. """ request = self._take(request_id) if request is None: return [] # An approval that arrives after the request went stale must not # put a ten-minute-old call on the screen. if self._too_old(request, now): return [ self._status(request.id, request.room, contract.EXPIRED), self._pending(), ] return self._start_delivery(request) + [self._pending()] def reject(self, request_id, now): request = self._take(request_id) if request is None: return [] return [ self._status(request.id, request.room, contract.REJECTED), self._pending(), ] def _take(self, request_id): for index, queued in enumerate(self.queue): if queued.id == request_id: return self.queue.pop(index) return None def tick(self, now): """Age the queue. Runs on its own timer, so requests keep expiring while the broker is away.""" stale = [held for held in self.queue if self._too_old(held, now)] if not stale: return [] self.queue = [held for held in self.queue if not self._too_old(held, now)] effects = [ self._status(held.id, held.room, contract.EXPIRED) for held in stale ] return effects + [self._pending()] def delivered(self, request, serial, ok, detail, now): """What ProPresenter said. A successful API call means the sanctuary accepted the message. It does not mean anybody saw it, so the strongest thing we say is "sent to sanctuary". """ if not ok: return [self._status(request.id, request.room, contract.ERROR, detail)] effects = [self._status(request.id, request.room, contract.SENT_TO_SANCTUARY)] if self.settings.propresenter_clear_after > 0: effects.append( ScheduleClear(serial, self.settings.propresenter_clear_after) ) return effects def clear_due(self, serial): """A clear timer fired. Only the newest delivery may clear the screen: an older timer firing after a second call has gone up must not wipe it. """ if serial != self.serial: return [] return [ClearNow(serial)]