factor out notification logic into Notifier class

chatmaild/src/chatmaild/metadata.py

@@ -1,4 +1,6 @@
 import pwd
+
+from queue import Queue
 from socketserver import (
     UnixStreamServer,
     StreamRequestHandler,
@@ -18,7 +20,35 @@ DICTPROXY_COMMIT_TRANSACTION_CHAR = "C"
 DICTPROXY_TRANSACTION_CHARS = "SBC"
 
 
-def handle_dovecot_protocol(rfile, wfile, tokens, notify_guid):
+class Notifier:
+    def __init__(self):
+        self.guid2token = {}
+        self.to_notify_queue = Queue()
+
+    def set_token(self, guid, token):
+        self.guid2token[guid] = token
+
+    def new_message_for_guid(self, guid):
+        self.to_notify_queue.put(guid)
+
+    def thread_run(self):
+        requests_session = requests.Session()
+        while 1:
+            guid = self.to_notify_queue.get()
+            token = self.guid2token.get(guid)
+            if token:
+                response = requests_session.post(
+                    "https://notifications.delta.chat/notify",
+                    data=token,
+                    timeout=60,
+                )
+                if response.status_code == 410:
+                    # 410 Gone status code
+                    # means the token is no longer valid.
+                    del self.guid2token[guid]
+
+
+def handle_dovecot_protocol(rfile, wfile, notifier):
     # HELLO message, ignored.
     msg = rfile.readline().strip().decode()
 
@@ -28,13 +58,13 @@ def handle_dovecot_protocol(rfile, wfile, tokens, notify_guid):
         if not msg:
             break
 
-        res = handle_dovecot_request(msg, transactions, tokens, notify_guid)
+        res = handle_dovecot_request(msg, transactions, notifier)
         if res:
             wfile.write(res.encode("ascii"))
             wfile.flush()
 
 
-def handle_dovecot_request(msg, transactions, tokens, notify_guid):
+def handle_dovecot_request(msg, transactions, notifier):
     # see https://doc.dovecot.org/3.0/developer_manual/design/dict_protocol/
     print("got", msg)
     short_command = msg[0]
@@ -68,10 +98,9 @@ def handle_dovecot_request(msg, transactions, tokens, notify_guid):
         keyname = parts[1].split("/")
         value = parts[2] if len(parts) > 2 else ""
         if keyname[0] == "priv" and keyname[2] == "devicetoken":
-            tokens[keyname[1]] = value
+            notifier.set_token(keyname[1], value)
         elif keyname[0] == "priv" and keyname[2] == "messagenew":
-            guid = keyname[1]
+            notifier.new_message_for_guid(keyname[1])
-            notify_guid(guid)
         else:
             # Transaction failed.
             transactions[transaction_id] = "F\n"
@@ -87,28 +116,12 @@ def main():
 
     # XXX config is not currently used
     config = read_config(config)
-    tokens = {}
+    notifier = Notifier()
-    requests_session = requests.Session()
-
-    def notify_guid(guid):
-        token = tokens.get(guid)
-        if token:
-            response = requests_session.post(
-                "https://notifications.delta.chat/notify",
-                data=tokens[guid],
-                timeout=60,
-            )
-            if response.status_code == 410:
-                # 410 Gone status code
-                # means the token is no longer valid.
-                del tokens[guid]
 
     class Handler(StreamRequestHandler):
         def handle(self):
             try:
-                handle_dovecot_protocol(
+                handle_dovecot_protocol(self.rfile, self.wfile, notifier)
-                    self.rfile, self.wfile, tokens, requests_session
-                )
             except Exception:
                 logging.exception("Exception in the handler")
                 raise

chatmaild/src/chatmaild/tests/test_metadata.py

@@ -3,49 +3,52 @@ import io
 from chatmaild.metadata import (
     handle_dovecot_request,
     handle_dovecot_protocol,
+    Notifier,
 )
 
 
 def test_handle_dovecot_request_lookup_fails():
-    res = handle_dovecot_request("Lpriv/123/chatmail", {}, {}, None)
+    notifier = Notifier()
+    res = handle_dovecot_request("Lpriv/123/chatmail", {}, notifier)
     assert res == "N\n"
 
 
 def test_handle_dovecot_request_happy_path():
-    tokens = {}
+    notifier = Notifier()
     transactions = {}
 
     # lookups return the same NOTFOUND result
-    res = handle_dovecot_request("Lpriv/123/chatmail", transactions, tokens, None)
+    res = handle_dovecot_request("Lpriv/123/chatmail", transactions, notifier)
     assert res == "N\n"
-    assert not tokens and not transactions
+    assert not notifier.guid2token and not transactions
 
     # set device token in a transaction
     tx = "1111"
     msg = f"B{tx}\tuser"
-    res = handle_dovecot_request(msg, transactions, tokens, None)
+    res = handle_dovecot_request(msg, transactions, notifier)
-    assert not res and not tokens
+    assert not res and not notifier.guid2token
     assert transactions == {tx: "O\n"}
 
     msg = f"S{tx}\tpriv/guid00/devicetoken\t01234"
-    res = handle_dovecot_request(msg, transactions, tokens, None)
+    res = handle_dovecot_request(msg, transactions, notifier)
     assert not res
     assert len(transactions) == 1
-    assert len(tokens) == 1 and tokens["guid00"] == "01234"
+    assert len(notifier.guid2token) == 1
+    assert notifier.guid2token["guid00"] == "01234"
 
     msg = f"C{tx}"
-    res = handle_dovecot_request(msg, transactions, tokens, None)
+    res = handle_dovecot_request(msg, transactions, notifier)
     assert res == "O\n"
     assert len(transactions) == 0
-    assert tokens["guid00"] == "01234"
+    assert notifier.guid2token["guid00"] == "01234"
 
     # trigger notification for incoming message
-    assert handle_dovecot_request(f"B{tx}\tuser", transactions, tokens, None) is None
+    assert handle_dovecot_request(f"B{tx}\tuser", transactions, notifier) is None
     msg = f"S{tx}\tpriv/guid00/messagenew"
-    requests = []
+    assert handle_dovecot_request(msg, transactions, notifier) is None
-    assert handle_dovecot_request(msg, transactions, tokens, requests.append) is None
+    assert notifier.to_notify_queue.get() == "guid00"
-    assert requests == ["guid00"]
+    assert notifier.to_notify_queue.qsize() == 0
-    assert handle_dovecot_request(f"C{tx}\tuser", transactions, tokens, None) == "O\n"
+    assert handle_dovecot_request(f"C{tx}\tuser", transactions, notifier) == "O\n"
     assert not transactions
 
 
@@ -60,10 +63,10 @@ def test_handle_dovecot_protocol_set_devicetoken():
             ]
         )
     )
-    tokens = {}
     wfile = io.BytesIO()
-    handle_dovecot_protocol(rfile, wfile, tokens, None)
+    notifier = Notifier()
-    assert tokens["guid00"] == "01234"
+    handle_dovecot_protocol(rfile, wfile, notifier)
+    assert notifier.guid2token["guid00"] == "01234"
     assert wfile.getvalue() == b"O\n"
 
 
@@ -78,9 +81,9 @@ def test_handle_dovecot_protocol_messagenew():
             ]
         )
     )
-    tokens = {"guid00": "01234"}
     wfile = io.BytesIO()
-    requests = []
+    notifier = Notifier()
-    handle_dovecot_protocol(rfile, wfile, tokens, requests.append)
+    handle_dovecot_protocol(rfile, wfile, notifier)
     assert wfile.getvalue() == b"O\n"
-    assert requests == ["guid00"]
+    assert notifier.to_notify_queue.get() == "guid00"
+    assert notifier.to_notify_queue.qsize() == 0