message queues

nostr/client/client.py

@@ -21,10 +21,9 @@ from . import cbc
 
 class NostrClient:
     relays = [
-        # "wss://lnbits.link/nostrrelay/client"
-        "wss://nostr-pub.wellorder.net",
-        # "wss://nostr.zebedee.cloud",
-        # "wss://no.str.cr",
+        "wss://lnbits.link/nostrrelay/client" "wss://nostr-pub.wellorder.net",
+        "wss://nostr.zebedee.cloud",
+        "wss://no.str.cr",
     ]  # ["wss://nostr.oxtr.dev"]  # ["wss://relay.nostr.info"] "wss://nostr-pub.wellorder.net"  "ws://91.237.88.218:2700", "wss://nostrrr.bublina.eu.org", ""wss://nostr-relay.freeberty.net"", , "wss://nostr.oxtr.dev", "wss://relay.nostr.info", "wss://nostr-pub.wellorder.net" , "wss://relayer.fiatjaf.com", "wss://nodestr.fmt.wiz.biz/", "wss://no.str.cr"
     relay_manager = RelayManager()
     private_key: PrivateKey
@@ -44,6 +43,7 @@ class NostrClient:
         self.relay_manager.open_connections(
             {"cert_reqs": ssl.CERT_NONE}
         )  # NOTE: This disables ssl certificate verification
+        self.relay_manager.start_message_workers()
 
     def close(self):
         self.relay_manager.close_connections()

nostr/relay.py

@@ -1,5 +1,6 @@
 import json
 import time
+from queue import Queue
 from threading import Lock
 from websocket import WebSocketApp
 from .event import Event
@@ -37,6 +38,7 @@ class Relay:
         self.ssl_options: dict = {}
         self.proxy: dict = {}
         self.lock = Lock()
+        self.queue = Queue()
         self.ws = WebSocketApp(
             url,
             on_open=self._on_open,
@@ -69,8 +71,15 @@ class Relay:
             self.connect(self.ssl_options, self.proxy)
 
     def publish(self, message: str):
-        if self.connected:
-            self.ws.send(message)
+        self.queue.put(message)
+
+    def queue_worker(self):
+        while True:
+            if self.connected:
+                message = self.queue.get()
+                self.ws.send(message)
+            else:
+                time.sleep(0.1)
 
     def add_subscription(self, id, filters: Filters):
         with self.lock:

nostr/relay_manager.py

@@ -43,6 +43,13 @@ class RelayManager:
                 name=f"{relay.url}-thread",
             ).start()
 
+    def start_message_workers(self):
+        for relay in self.relays.values():
+            threading.Thread(
+                target=relay.queue_worker,
+                name=f"{relay.url}-queue",
+            ).start()
+
     def close_connections(self):
         for relay in self.relays.values():
             relay.close()