This commit is contained in:
Nicolas Mowen 2024-02-14 07:31:33 -07:00
parent a3fb1325b2
commit ed1cd052db

View File

@ -2,7 +2,6 @@
import multiprocessing as mp import multiprocessing as mp
import os import os
import threading
from multiprocessing.synchronize import Event as MpEvent from multiprocessing.synchronize import Event as MpEvent
from typing import Callable, Optional from typing import Callable, Optional
@ -30,28 +29,10 @@ class ConfigPublisher(Communicator):
self.socket.send_pyobj(payload) self.socket.send_pyobj(payload)
def subscribe(self, receiver: Callable) -> None: def subscribe(self, receiver: Callable) -> None:
self._dispatcher = receiver pass # this class does not subscribe
self.reader_thread = threading.Thread(target=self.read)
self.reader_thread.start()
def read(self) -> None:
while not self.stop_event.wait(0.5):
while True: # load all messages that are queued
try:
(topic, value) = self.socket.recv_pyobj(flags=zmq.NOBLOCK)
response = self._dispatcher(topic, value)
if response is not None:
self.socket.send_pyobj(response)
else:
self.socket.send_pyobj([])
except zmq.ZMQError:
break
def stop(self) -> None: def stop(self) -> None:
self.stop_event.set() self.stop_event.set()
self.reader_thread.join()
self.socket.close() self.socket.close()
self.context.destroy() self.context.destroy()