Ruim stale account topics op bij startup

plugins/accounts.py

@@ -5,6 +5,8 @@
 import time
 import codecs
 import logging
+import paho.mqtt.client as mqtt
+from config import config_get
 
 logger = logging.getLogger(__name__)
 
@@ -128,6 +130,48 @@ def publish_members(self):
     def _publish_members(self):
         self.publish_members()
 
+    def _account_topic(self, name):
+        return "accounts/" + name
+
+    def _publish_account(self, name, account):
+        self.master.send_message(True, self._account_topic(name), json.dumps(account))
+
+    def _clear_account_topic(self, name):
+        self.master.send_message(True, self._account_topic(name), "")
+
+    def _retained_account_names(self):
+        names = set()
+        topic_prefix = "hack42bar/output/session/" + self.SID + "/accounts/"
+
+        def on_message(_client, _userdata, msg):
+            if msg.payload:
+                names.add(msg.topic[len(topic_prefix) :])
+
+        try:
+            mqtt_config = config_get("mqtt", default={})
+            client = mqtt.Client()
+            client.on_message = on_message
+            client.connect(
+                mqtt_config["host"],
+                int(mqtt_config["port"]),
+                int(mqtt_config["keepalive"]),
+            )
+            client.subscribe(topic_prefix + "+")
+            deadline = time.monotonic() + float(
+                mqtt_config.get("retained_scan_timeout", 0.5)
+            )
+            while time.monotonic() < deadline:
+                if client.loop(timeout=0.05) != mqtt.MQTT_ERR_SUCCESS:
+                    break
+            client.disconnect()
+        except (KeyError, OSError, TypeError, ValueError):
+            logger.exception("retained_account_scan_failed sid=%s", self.SID)
+        return names
+
+    def _clear_removed_account_topics(self):
+        for name in sorted(self._retained_account_names() - set(self.accounts)):
+            self._clear_account_topic(name)
+
     def updateaccount(self, usr, value):
         logger.debug("update_account sid=%s user=%s value=%s", self.SID, usr, value)
         if usr == "cash":
@@ -165,9 +209,7 @@ def hook_balance(self, args):
         self.master.send_message(
             False, "infobox/account/" + usr, json.dumps(self.accounts[usr])
         )
-        self.master.send_message(
-            True, "accounts/" + usr, json.dumps(self.accounts[usr])
-        )
+        self._publish_account(usr, self.accounts[usr])
 
     def hook_endsession(self, _text):
         self.writeaccount()
@@ -177,7 +219,7 @@ def hook_abort(self, _void):
         self.readaccounts()
         self.get_last_updated_accounts()
         for name, account in self.accounts.items():
-            self.master.send_message(True, "accounts/" + name, json.dumps(account))
+            self._publish_account(name, account)
 
     def createnew(self, text):
         if text == "yes":
@@ -211,9 +253,10 @@ def createnew(self, text):
     def startup(self):
         self.readaccounts()
         self.readmembers()
+        self._clear_removed_account_topics()
         self.get_last_updated_accounts()
         for name, account in self.accounts.items():
-            self.master.send_message(True, "accounts/" + name, json.dumps(account))
+            self._publish_account(name, account)
         self.master.send_message(True, "members", json.dumps(self.visible_members()))
 
     def hook_pre_checkout(self, _text):

tests/js/kassa-app.test.js

@@ -18,8 +18,18 @@ function makeElement(tag, register) {
     scrollTop: 0,
     style: {},
     tag,
-    textContent: "",
+    _textContent: "",
     value: "",
+    get textContent() {
+      return this._textContent;
+    },
+    set textContent(value) {
+      this._textContent = String(value);
+      if(value === "") {
+        this.children = [];
+        this.childElementCount = 0;
+      }
+    },
     appendChild(child) {
       child.parentElement = this;
       this.children.push(child);
@@ -279,3 +289,31 @@ test("kassa app handles closed streams by scheduling one reconnect", () => {
   assert.equal(timers.length, 1);
   assert.equal(timers[0].delay, 1000);
 });
+
+test("kassa app removes accounts when retained account messages are cleared", () => {
+  const {elements, eventSources} = loadKassaApp();
+  const sendStreamMessage = (topic, msg) => {
+    eventSources[0].onmessage({data: JSON.stringify([topic, msg])});
+  };
+  const accountButtonTexts = () => (
+    findAll(elements.MainButtons, ".Buttontext").map((element) => element.textContent)
+  );
+
+  sendStreamMessage(
+    "hack42bar/output/session/main/accounts/user1",
+    '{"amount": 1, "lastupdate": "now"}',
+  );
+  sendStreamMessage(
+    "hack42bar/output/session/main/accounts/stale",
+    '{"amount": 2, "lastupdate": "now"}',
+  );
+  sendStreamMessage("hack42bar/output/session/main/members", '["user1", "stale"]');
+  sendStreamMessage("hack42bar/output/session/main/buttons", '{"special": "accounts"}');
+
+  assert.deepEqual(accountButtonTexts(), ["stale", "user1"]);
+
+  sendStreamMessage("hack42bar/output/session/main/accounts/stale", "");
+  sendStreamMessage("hack42bar/output/session/main/buttons", '{"special": "accounts"}');
+
+  assert.deepEqual(accountButtonTexts(), ["user1"]);
+});

tests/plugins/test_accounts.py

@@ -207,6 +207,90 @@ def test_visible_members_falls_back_to_members_file():
     assert acc.visible_members() == ["user1"]
 
 
+def test_retained_account_names_reads_existing_mqtt_topics():
+    class FakeMessage:
+        def __init__(self, topic, payload):
+            self.topic = topic
+            self.payload = payload
+
+    class FakeMqttClient:
+        def __init__(self):
+            self.on_message = None
+            self.connected = None
+            self.subscribed = None
+            self.disconnected = False
+
+        def connect(self, host, port, keepalive):
+            self.connected = (host, port, keepalive)
+
+        def subscribe(self, topic):
+            self.subscribed = topic
+
+        def loop(self, timeout=0):
+            assert timeout == 0.05
+            self.on_message(
+                self,
+                None,
+                FakeMessage("hack42bar/output/session/SID/accounts/user1", b"{}"),
+            )
+            self.on_message(
+                self,
+                None,
+                FakeMessage("hack42bar/output/session/SID/accounts/stale", b"{}"),
+            )
+            self.on_message(
+                self,
+                None,
+                FakeMessage("hack42bar/output/session/SID/accounts/empty", b""),
+            )
+            return 1
+
+        def disconnect(self):
+            self.disconnected = True
+
+    fake_client = FakeMqttClient()
+    master_mock = Mock()
+    acc = accounts("SID", master_mock)
+
+    with patch("plugins.accounts.mqtt.Client", return_value=fake_client), patch(
+        "plugins.accounts.config_get",
+        return_value={"host": "mqtt.example.test", "port": 1884, "keepalive": 30},
+    ):
+        assert acc._retained_account_names() == {"user1", "stale"}
+
+    assert fake_client.connected == ("mqtt.example.test", 1884, 30)
+    assert fake_client.subscribed == "hack42bar/output/session/SID/accounts/+"
+    assert fake_client.disconnected is True
+
+
+def test_retained_account_names_logs_scan_failures(caplog):
+    class FailingMqttClient:
+        def connect(self, _host, _port, _keepalive):
+            raise OSError("mqtt down")
+
+    master_mock = Mock()
+    acc = accounts("SID", master_mock)
+
+    with patch("plugins.accounts.mqtt.Client", return_value=FailingMqttClient()), patch(
+        "plugins.accounts.config_get",
+        return_value={"host": "mqtt.example.test", "port": 1884, "keepalive": 30},
+    ):
+        assert acc._retained_account_names() == set()
+
+    assert "retained_account_scan_failed sid=SID" in caplog.text
+
+
+def test_clear_removed_account_topics_clears_stale_retained_topics():
+    master_mock = Mock()
+    acc = accounts("SID", master_mock)
+    acc.accounts = {"user1": {"amount": 0, "lastupdate": "2021-01-01"}}
+
+    with patch.object(acc, "_retained_account_names", return_value={"stale", "user1"}):
+        acc._clear_removed_account_topics()
+
+    master_mock.send_message.assert_called_once_with(True, "accounts/stale", "")
+
+
 def test_hook_balance():
     master_mock = Mock()
     acc = accounts("SID", master_mock)
@@ -349,7 +433,7 @@ def custom_mock_open(filename, _bla, _bla2):
 
     with patch(
         "plugins.accounts.codecs.open", side_effect=custom_mock_open
-    ) as mock_file:
+    ), patch.object(acc, "_retained_account_names", return_value=set()):
         acc.startup()
 
     assert acc.members == ["user1", "user2"]

www/kassa-app.js

@@ -198,6 +198,10 @@ document.addEventListener("DOMContentLoaded", function() {
     appendToElement(infobox, createElement('span',{class: 'NewSaldo',text: newsaldo.toFixed(2)}));
   }
   function setupaccounts(name,msg) {
+    if(msg === "") {
+      delete accounts[name];
+      return;
+    }
     accounts[name]=JSON.parse(msg);
   }
   function setupcommands(msg) {