EntitiesEngine synchronization improvements.

- Added `wait_start()` method that other threads can use to synchronize
  with the engine and wait before performing db operations.

- Callback logic wrapped in a try/except block to prevent custom
  integrations with buggy callbacks from crashing the engine.
This commit is contained in:
Fabio Manganiello 2023-03-10 00:57:24 +01:00
parent 73dc2463f1
commit ceb7a2f098
Signed by untrusted user: blacklight
GPG key ID: D90FBA7F76362774

View file

@ -37,6 +37,11 @@ class EntitiesEngine(Thread):
self.logger = getLogger(name=obj_name) self.logger = getLogger(name=obj_name)
self._should_stop = Event() self._should_stop = Event()
""" Event used to synchronize stop events downstream.""" """ Event used to synchronize stop events downstream."""
self._running = Event()
"""
Event used to synchronize other threads to wait for the engine to
start.
"""
self._queue = EntitiesQueue(stop_event=self._should_stop) self._queue = EntitiesQueue(stop_event=self._should_stop)
""" Queue where all entity upsert requests are received.""" """ Queue where all entity upsert requests are received."""
self._repo = EntitiesRepository() self._repo = EntitiesRepository()
@ -51,6 +56,13 @@ class EntitiesEngine(Thread):
self._queue.put(*entities) self._queue.put(*entities)
def wait_start(self, timeout: Optional[float] = None) -> None:
started = self._running.wait(timeout=timeout)
if not started:
raise TimeoutError(
f'Timeout waiting for {self.__class__.__name__} to start.'
)
@property @property
def should_stop(self) -> bool: def should_stop(self) -> bool:
return self._should_stop.is_set() return self._should_stop.is_set()
@ -66,15 +78,32 @@ class EntitiesEngine(Thread):
""" """
for entity in entities: for entity in entities:
get_bus().post(EntityUpdateEvent(entity=entity)) get_bus().post(EntityUpdateEvent(entity=entity))
self._process_callback(entity)
def _process_callback(self, entity: Entity) -> None:
"""
Process the callback for the given entity.
"""
callback = self._callbacks.pop(entity.entity_key, None) callback = self._callbacks.pop(entity.entity_key, None)
if callback: if callback:
try:
callback(entity) callback(entity)
except Exception as e:
self.logger.error(
'Error while notifying updates for entity ID %d via %s: %s',
entity.id,
callback,
e,
)
self.logger.exception(e)
def run(self): def run(self):
super().run() super().run()
set_thread_name('entities') set_thread_name('entities')
self.logger.info('Started entities engine') self.logger.info('Started entities engine')
self._running.set()
try:
while not self.should_stop: while not self.should_stop:
# Get a batch of entity updates forwarded by other integrations # Get a batch of entity updates forwarded by other integrations
entities = self._queue.get() entities = self._queue.get()
@ -91,5 +120,6 @@ class EntitiesEngine(Thread):
# Trigger EntityUpdateEvent events # Trigger EntityUpdateEvent events
self.notify(*entities) self.notify(*entities)
finally:
self.logger.info('Stopped entities engine') self.logger.info('Stopped entities engine')
self._running.clear()