Git-Mirrors/NomadNet
1
0
Fork 0
mirror of https://github.com/markqvist/NomadNet.git synced 2025-08-19 11:57:57 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

Use lock on announce stream updates

Browse source
This commit is contained in:
Mark Qvist 2025-01-28 23:04:22 +01:00
parent 9ef34fc774
commit eeb15dcb43
1 changed files with 80 additions and 75 deletions
Download patch file Download diff file Expand all files Collapse all files

155
nomadnet/Directory.py
View file

@ -3,6 +3,7 @@ import RNS
import LXMF import LXMF
import time import time
import nomadnet import nomadnet
import threading
import RNS.vendor.umsgpack as msgpack import RNS.vendor.umsgpack as msgpack
class PNAnnounceHandler: class PNAnnounceHandler:
@ -55,6 +56,7 @@ class Directory:
self.directory_entries = {} self.directory_entries = {}
self.announce_stream = [] self.announce_stream = []
self.app = app self.app = app
self.announce_lock = threading.Lock()
self.load_from_disk() self.load_from_disk()
self.pn_announce_handler = PNAnnounceHandler(self) self.pn_announce_handler = PNAnnounceHandler(self)
@ -124,91 +126,94 @@ class Directory:
RNS.log("Could not load directory from disk. The contained exception was: "+str(e), RNS.LOG_ERROR) RNS.log("Could not load directory from disk. The contained exception was: "+str(e), RNS.LOG_ERROR)
def lxmf_announce_received(self, source_hash, app_data): def lxmf_announce_received(self, source_hash, app_data):
if app_data != None: with self.announce_lock:
if self.app.compact_stream: if app_data != None:
try: if self.app.compact_stream:
remove_announces = [] try:
for announce in self.announce_stream: remove_announces = []
if announce[1] == source_hash: for announce in self.announce_stream:
remove_announces.append(announce) if announce[1] == source_hash:
remove_announces.append(announce)
for a in remove_announces: for a in remove_announces:
self.announce_stream.remove(a) self.announce_stream.remove(a)
except Exception as e:
RNS.log("An error occurred while compacting the announce stream. The contained exception was:"+str(e), RNS.LOG_ERROR)
timestamp = time.time()
self.announce_stream.insert(0, (timestamp, source_hash, app_data, "peer"))
while len(self.announce_stream) > Directory.ANNOUNCE_STREAM_MAXLENGTH:
self.announce_stream.pop()
if hasattr(self.app, "ui") and self.app.ui != None:
if hasattr(self.app.ui, "main_display"):
self.app.ui.main_display.sub_displays.network_display.directory_change_callback()
def node_announce_received(self, source_hash, app_data, associated_peer):
with self.announce_lock:
if app_data != None:
if self.app.compact_stream:
try:
remove_announces = []
for announce in self.announce_stream:
if announce[1] == source_hash:
remove_announces.append(announce)
for a in remove_announces:
self.announce_stream.remove(a)
except Exception as e:
RNS.log("An error occurred while compacting the announce stream. The contained exception was:"+str(e), RNS.LOG_ERROR)
timestamp = time.time()
self.announce_stream.insert(0, (timestamp, source_hash, app_data, "node"))
while len(self.announce_stream) > Directory.ANNOUNCE_STREAM_MAXLENGTH:
self.announce_stream.pop()
if self.trust_level(associated_peer) == DirectoryEntry.TRUSTED:
existing_entry = self.find(source_hash)
if not existing_entry:
node_entry = DirectoryEntry(source_hash, display_name=app_data.decode("utf-8"), trust_level=DirectoryEntry.TRUSTED, hosts_node=True)
self.remember(node_entry)
except Exception as e:
RNS.log("An error occurred while compacting the announce stream. The contained exception was:"+str(e), RNS.LOG_ERROR)
timestamp = time.time()
self.announce_stream.insert(0, (timestamp, source_hash, app_data, "peer"))
while len(self.announce_stream) > Directory.ANNOUNCE_STREAM_MAXLENGTH:
self.announce_stream.pop()
if hasattr(self.app, "ui") and self.app.ui != None:
if hasattr(self.app.ui, "main_display"): if hasattr(self.app.ui, "main_display"):
self.app.ui.main_display.sub_displays.network_display.directory_change_callback() self.app.ui.main_display.sub_displays.network_display.directory_change_callback()
def node_announce_received(self, source_hash, app_data, associated_peer):
if app_data != None:
if self.app.compact_stream:
try:
remove_announces = []
for announce in self.announce_stream:
if announce[1] == source_hash:
remove_announces.append(announce)
for a in remove_announces:
self.announce_stream.remove(a)
except Exception as e:
RNS.log("An error occurred while compacting the announce stream. The contained exception was:"+str(e), RNS.LOG_ERROR)
timestamp = time.time()
self.announce_stream.insert(0, (timestamp, source_hash, app_data, "node"))
while len(self.announce_stream) > Directory.ANNOUNCE_STREAM_MAXLENGTH:
self.announce_stream.pop()
if self.trust_level(associated_peer) == DirectoryEntry.TRUSTED:
existing_entry = self.find(source_hash)
if not existing_entry:
node_entry = DirectoryEntry(source_hash, display_name=app_data.decode("utf-8"), trust_level=DirectoryEntry.TRUSTED, hosts_node=True)
self.remember(node_entry)
if hasattr(self.app.ui, "main_display"):
self.app.ui.main_display.sub_displays.network_display.directory_change_callback()
def pn_announce_received(self, source_hash, app_data, associated_peer, associated_node): def pn_announce_received(self, source_hash, app_data, associated_peer, associated_node):
found_node = None with self.announce_lock:
for sh in self.directory_entries: found_node = None
if sh == associated_node: for sh in self.directory_entries:
found_node = True if sh == associated_node:
break found_node = True
break
for e in self.announce_stream: for e in self.announce_stream:
if e[1] == associated_node: if e[1] == associated_node:
found_node = True found_node = True
break break
if not found_node: if not found_node:
if self.app.compact_stream: if self.app.compact_stream:
try: try:
remove_announces = [] remove_announces = []
for announce in self.announce_stream: for announce in self.announce_stream:
if announce[1] == source_hash: if announce[1] == source_hash:
remove_announces.append(announce) remove_announces.append(announce)
for a in remove_announces: for a in remove_announces:
self.announce_stream.remove(a) self.announce_stream.remove(a)
except Exception as e:
RNS.log("An error occurred while compacting the announce stream. The contained exception was:"+str(e), RNS.LOG_ERROR)
timestamp = time.time()
self.announce_stream.insert(0, (timestamp, source_hash, app_data, "pn"))
while len(self.announce_stream) > Directory.ANNOUNCE_STREAM_MAXLENGTH:
self.announce_stream.pop()
except Exception as e: if hasattr(self.app.ui, "main_display"):
RNS.log("An error occurred while compacting the announce stream. The contained exception was:"+str(e), RNS.LOG_ERROR) self.app.ui.main_display.sub_displays.network_display.directory_change_callback()
timestamp = time.time()
self.announce_stream.insert(0, (timestamp, source_hash, app_data, "pn"))
while len(self.announce_stream) > Directory.ANNOUNCE_STREAM_MAXLENGTH:
self.announce_stream.pop()
if hasattr(self.app.ui, "main_display"):
self.app.ui.main_display.sub_displays.network_display.directory_change_callback()
def remove_announce_with_timestamp(self, timestamp): def remove_announce_with_timestamp(self, timestamp):
selected_announce = None selected_announce = None