Changes from callebtc#8

Author: vic

main.py

@@ -27,7 +27,7 @@ def callback(event: Event, decrypted_content):
             f"\nFrom {event.public_key[:3]}..{event.public_key[-3:]}: {decrypted_content}"
         )
 
-    client = NostrClient(private_key=pk)
+    client = NostrClient(privatekey_hex=pk)
     if not pk:
         print(f"Your private key: {client.private_key.bech32()}")
 
@@ -66,7 +66,7 @@ def callback(event: Event):
             f"\nFrom {event.public_key[:3]}..{event.public_key[-3:]}: {event.content}"
         )
 
-    sender_client = NostrClient(private_key=pk)
+    sender_client = NostrClient(privatekey_hex=pk)
     # await asyncio.sleep(1)
 
     pubkey_to_str = (

nostr/bech32.py

@@ -23,6 +23,7 @@
 
 from enum import Enum
 
+
 class Encoding(Enum):
     """Enumeration type to list the various supported encodings."""
     BECH32 = 1

nostr/client/client.py

@@ -1,44 +1,33 @@
-from typing import *
-import ssl
-import time
+import base64
 import json
 import os
-import base64
-
-from ..event import Event
-from ..relay_manager import RelayManager
-from ..message_type import ClientMessageType
-from ..key import PrivateKey, PublicKey
+import ssl
+import time
+from typing import *
 
+from ..event import EncryptedDirectMessage, Event, EventKind
 from ..filter import Filter, Filters
-from ..event import Event, EventKind, EncryptedDirectMessage
-from ..relay_manager import RelayManager
+from ..key import PrivateKey, PublicKey
 from ..message_type import ClientMessageType
+from ..relay_manager import RelayManager
 
 # from aes import AESCipher
 from . import cbc
 
 
 class NostrClient:
     relays = [
-        # "wss://eagerporpoise9.lnbits.com/nostrclient/api/v1/relay",
-        "wss://localhost:5001/nostrclient/api/v1/relay",
-        # "wss://nostr-pub.wellorder.net",
-        # "wss://relay.damus.io",
-        # "wss://nostr.zebedee.cloud",
-        # "wss://relay.snort.social",
-        # "wss://nostr.fmt.wiz.biz",
-        # "wss://nos.lol",
-        # "wss://nostr.oxtr.dev",
-        # "wss://relay.current.fyi",
-        # "wss://relay.snort.social",
+        "wss://nostr-pub.wellorder.net",
+        "wss://nostr.zebedee.cloud",
+        "wss://nodestr.fmt.wiz.biz",
+        "wss://nostr.oxtr.dev",        
     ]  # ["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
     public_key: PublicKey
 
-    def __init__(self, private_key: str = "", relays: List[str] = [], connect=True):
-        self.generate_keys(private_key)
+    def __init__(self, privatekey_hex: str = "", relays: List[str] = [], connect=True):
+        self.generate_keys(privatekey_hex)
 
         if len(relays):
             self.relays = relays
@@ -55,13 +44,9 @@ def connect(self):
     def close(self):
         self.relay_manager.close_connections()
 
-    def generate_keys(self, private_key: str = None):
-        if private_key.startswith("nsec"):
-            self.private_key = PrivateKey.from_nsec(private_key)
-        elif private_key:
-            self.private_key = PrivateKey(bytes.fromhex(private_key))
-        else:
-            self.private_key = PrivateKey()  # generate random key
+    def generate_keys(self, privatekey_hex: str = None):
+        pk = bytes.fromhex(privatekey_hex) if privatekey_hex else None
+        self.private_key = PrivateKey(pk)
         self.public_key = self.private_key.public_key
 
     def post(self, message: str):
@@ -87,7 +72,6 @@ def get_post(
         request = [ClientMessageType.REQUEST, subscription_id]
         request.extend(filters.to_json_array())
         message = json.dumps(request)
-        # print(message)
         self.relay_manager.publish_message(message)
 
         while True:
@@ -102,16 +86,14 @@ def dm(self, message: str, to_pubkey: PublicKey):
             recipient_pubkey=to_pubkey.hex(), cleartext_content=message
         )
         self.private_key.sign_event(dm)
-        # print(dm)
         self.relay_manager.publish_event(dm)
 
-    def get_dm(self, sender_publickey: PublicKey, callback_func=None, filter_kwargs={}):
+    def get_dm(self, sender_publickey: PublicKey, callback_func=None):
         filters = Filters(
             [
                 Filter(
                     kinds=[EventKind.ENCRYPTED_DIRECT_MESSAGE],
                     pubkey_refs=[sender_publickey.hex()],
-                    **filter_kwargs,
                 )
             ]
         )
@@ -122,7 +104,7 @@ def get_dm(self, sender_publickey: PublicKey, callback_func=None, filter_kwargs=
         request.extend(filters.to_json_array())
         message = json.dumps(request)
         self.relay_manager.publish_message(message)
-        # print(message)
+
         while True:
             while self.relay_manager.message_pool.has_events():
                 event_msg = self.relay_manager.message_pool.get_event()
@@ -156,7 +138,7 @@ def subscribe(
                 if callback_events_func:
                     callback_events_func(event_msg)
             while self.relay_manager.message_pool.has_notices():
-                event_msg = self.relay_manager.message_pool.has_notices()
+                event_msg = self.relay_manager.message_pool.get_notice()
                 if callback_notices_func:
                     callback_notices_func(event_msg)
             while self.relay_manager.message_pool.has_eose_notices():

nostr/event.py

@@ -1,10 +1,11 @@
-import time
 import json
+import time
 from dataclasses import dataclass, field
 from enum import IntEnum
+from hashlib import sha256
 from typing import List
+
 from secp256k1 import PublicKey
-from hashlib import sha256
 
 from .message_type import ClientMessageType
 

nostr/key.py

@@ -1,14 +1,16 @@
-import secrets
 import base64
+import secrets
+from hashlib import sha256
+
 import secp256k1
 from cffi import FFI
-from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+
 from cryptography.hazmat.primitives import padding
-from hashlib import sha256
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
 
+from . import bech32
 from .delegation import Delegation
 from .event import EncryptedDirectMessage, Event, EventKind
-from . import bech32
 
 
 class PublicKey:

nostr/message_pool.py

@@ -1,8 +1,9 @@
 import json
 from queue import Queue
 from threading import Lock
-from .message_type import RelayMessageType
+
 from .event import Event
+from .message_type import RelayMessageType
 
 
 class EventMessage:

nostr/pow.py

@@ -1,7 +1,9 @@
 import time
+
 from .event import Event
 from .key import PrivateKey
 
+
 def zero_bits(b: int) -> int:
     n = 0
 

nostr/relay.py

@@ -2,7 +2,9 @@
 import time
 from queue import Queue
 from threading import Lock
-from websocket import WebSocketApp, WebSocketConnectionClosedException
+
+from websocket import WebSocketApp
+
 from .event import Event
 from .filter import Filters
 from .message_pool import MessagePool
@@ -33,26 +35,27 @@ def __init__(
         self.subscriptions = subscriptions
         self.connected: bool = False
         self.reconnect: bool = True
+        self.shutdown: bool = False
         self.error_counter: int = 0
-        self.error_threshold: int = 0
+        self.error_threshold: int = 100
         self.num_received_events: int = 0
         self.num_sent_events: int = 0
         self.num_subscriptions: int = 0
         self.ssl_options: dict = {}
         self.proxy: dict = {}
         self.lock = Lock()
         self.queue = Queue()
+
+    def connect(self, ssl_options: dict = None, proxy: dict = None):
         self.ws = WebSocketApp(
-            url,
+            self.url,
             on_open=self._on_open,
             on_message=self._on_message,
             on_error=self._on_error,
             on_close=self._on_close,
             on_ping=self._on_ping,
             on_pong=self._on_pong,
         )
-
-    def connect(self, ssl_options: dict = None, proxy: dict = None):
         self.ssl_options = ssl_options
         self.proxy = proxy
         if not self.connected:
@@ -66,6 +69,7 @@ def connect(self, ssl_options: dict = None, proxy: dict = None):
 
     def close(self):
         self.ws.close()
+        self.shutdown = True
 
     def check_reconnect(self):
         try:
@@ -74,7 +78,7 @@ def check_reconnect(self):
             pass
         self.connected = False
         if self.reconnect:
-            time.sleep(1)
+            time.sleep(self.error_counter**2)
             self.connect(self.ssl_options, self.proxy)
 
     @property
@@ -85,15 +89,16 @@ def ping(self):
     def publish(self, message: str):
         self.queue.put(message)
 
-    def queue_worker(self):
+    def queue_worker(self, shutdown):
         while True:
             if self.connected:
-                message = self.queue.get()
-                self.num_sent_events += 1
                 try:
+                    message = self.queue.get(timeout=1)
                     self.ws.send(message)
-                except WebSocketConnectionClosedException as wscce:
-                    self._on_error(None, wscce)
+                    self.num_sent_events += 1
+                except:
+                    if shutdown():
+                        break
             else:
                 time.sleep(0.1)
 
@@ -126,6 +131,10 @@ def _on_open(self, class_obj):
 
     def _on_close(self, class_obj, status_code, message):
         self.connected = False
+        if self.error_threshold and self.error_counter > self.error_threshold:
+            pass
+        else:
+            self.check_reconnect()        
         pass
 
     def _on_message(self, class_obj, message: str):
@@ -136,10 +145,6 @@ def _on_message(self, class_obj, message: str):
     def _on_error(self, class_obj, error):
         self.connected = False
         self.error_counter += 1
-        if self.error_threshold and self.error_counter > self.error_threshold:
-            pass
-        else:
-            self.check_reconnect()
 
     def _on_ping(self, class_obj, message):
         return

nostr/relay_manager.py

@@ -15,6 +15,8 @@ class RelayException(Exception):
 class RelayManager:
     def __init__(self) -> None:
         self.relays: dict[str, Relay] = {}
+        self.threads: dict[str, threading.Thread] = {}
+        self.queue_threads: dict[str, threading.Thread] = {}        
         self.message_pool = MessagePool()
 
     def add_relay(
@@ -25,7 +27,10 @@ def add_relay(
         self.relays[url] = relay
 
     def remove_relay(self, url: str):
+        self.relays[url].close()
         self.relays.pop(url)
+        self.threads[url].join(timeout=1)
+        self.threads.pop(url)        
 
     def add_subscription(self, id: str, filters: Filters):
         for relay in self.relays.values():
@@ -37,16 +42,21 @@ def close_subscription(self, id: str):
 
     def open_connections(self, ssl_options: dict = None, proxy: dict = None):
         for relay in self.relays.values():
-            threading.Thread(
+            self.threads[relay.url] = threading.Thread(
                 target=relay.connect,
                 args=(ssl_options, proxy),
                 name=f"{relay.url}-thread",
                 daemon=True,
-            ).start()
+            )
+            self.threads[relay.url].start()
 
-            threading.Thread(
-                target=relay.queue_worker, name=f"{relay.url}-queue", daemon=True
-            ).start()
+            self.queue_threads[relay.url] = threading.Thread(
+                target=relay.queue_worker, 
+                args=(lambda: relay.shutdown,),
+                name=f"{relay.url}-queue",
+                daemon=True,
+            )
+            self.queue_threads[relay.url].start()
 
     def close_connections(self):
         for relay in self.relays.values():

nostr/subscription.py

@@ -1,5 +1,6 @@
 from .filter import Filters
 
+
 class Subscription:
     def __init__(self, id: str, filters: Filters=None) -> None:
         self.id = id