Initial commit WIP

Author: AttilaGombosER

hello/advertizer.py

@@ -117,14 +117,14 @@ def stop(self) -> None:
     def advertise(self, info: ServiceInfo | None = None) -> None:
         self._advertizer.advertise(info)
 
-    def schedule(self, info: ServiceInfo | None = None, interval: float = 10, one_shot: bool = False) -> None:
+    def schedule(self, info: ServiceInfo | None = None, interval: float = 60, one_shot: bool = False) -> None:
         if one_shot:
             self._timer.start(interval, self.advertise, [info])
             log.info('One-shot service advertisement scheduled', service=info, interval=interval)
         else:
-            def periodic_advertise() -> None:
-                self.advertise(info)
-                self._timer.restart()
-
-            self._timer.start(interval, periodic_advertise)
+            self._timer.start(interval, self._advertise_and_restart, [info])
             log.info('Periodic service advertisement scheduled', service=info, interval=interval)
+
+    def _advertise_and_restart(self, info: ServiceInfo | None = None) -> None:
+        self.advertise(info)
+        self._timer.restart()

hello/discoverer.py

@@ -38,10 +38,13 @@ def discover(self, query: ServiceQuery | None = None) -> None:
     def get_services(self) -> dict[str, ServiceInfo]:
         raise NotImplementedError()
 
-    def register(self, callback: OnDiscoveryEvent) -> None:
+    def register(self, handler: OnDiscoveryEvent) -> None:
         raise NotImplementedError()
 
-    def deregister(self, callback: OnDiscoveryEvent) -> None:
+    def deregister(self, handler: OnDiscoveryEvent) -> None:
+        raise NotImplementedError()
+
+    def get_handlers(self) -> list[OnDiscoveryEvent]:
         raise NotImplementedError()
 
 
@@ -52,8 +55,8 @@ def __init__(self, sender: Sender, receiver: Receiver) -> None:
         self._receiver = receiver
         self._group: Group | None = None
         self._matcher: ServiceMatcher | None = None
-        self._services: dict[str, ServiceInfo] = {}
-        self._callbacks: list[OnDiscoveryEvent] = []
+        self._cache: dict[str, ServiceInfo] = {}
+        self._handlers: list[OnDiscoveryEvent] = []
 
     def __enter__(self) -> Discoverer:
         return self
@@ -85,28 +88,27 @@ def discover(self, query: ServiceQuery | None = None) -> None:
             log.warning('Cannot discover services, discoverer not started', query=query)
 
     def get_services(self) -> dict[str, ServiceInfo]:
-        return self._services.copy()
+        return self._cache.copy()
 
-    def register(self, callback: OnDiscoveryEvent) -> None:
-        self._callbacks.append(callback)
+    def register(self, handler: OnDiscoveryEvent) -> None:
+        self._handlers.append(handler)
 
-    def deregister(self, callback: OnDiscoveryEvent) -> None:
-        self._callbacks.remove(callback)
+    def deregister(self, handler: OnDiscoveryEvent) -> None:
+        self._handlers.remove(handler)
 
-    def _handle_message(self, message: dict[str, Any]) -> None:
-        service: ServiceInfo | None = None
+    def get_handlers(self) -> list[OnDiscoveryEvent]:
+        return self._handlers.copy()
 
+    def _handle_message(self, message: dict[str, Any]) -> None:
         try:
             service = ServiceInfo(**message)
+            self._handle_service(service)
         except Exception as error:
             log.warn('Failed to handle received message', data=message, error=error)
 
-        if service:
-            self._handle_service(service)
-
     def _handle_service(self, service: ServiceInfo) -> None:
         if self._matcher and self._matcher.matches(service):
-            cached = self._services.get(service.name)
+            cached = self._cache.get(service.name)
 
             if event := self._create_event(cached, service):
                 self._handle_event(event)
@@ -116,17 +118,17 @@ def _create_event(self, cached: ServiceInfo | None, service: ServiceInfo) -> Dis
             if cached != service:
                 log.info('Service updated', old_service=cached, new_service=service)
                 return DiscoveryEvent(service, DiscoveryEventType.UPDATED)
+            else:
+                return None
         else:
             log.info('Service discovered', service=service)
             return DiscoveryEvent(service, DiscoveryEventType.DISCOVERED)
 
-        return None
-
     def _handle_event(self, event: DiscoveryEvent) -> None:
         service = event.service
-        self._services[service.name] = service
-        for callback in self._callbacks:
+        self._cache[service.name] = service
+        for callback in self._handlers:
             try:
                 callback(event)
             except Exception as error:
-                log.warn('Error in callback execution', service=service, error=error)
+                log.warn('Error in event handler execution', event=event, error=error)

hello/receiver.py

@@ -2,7 +2,7 @@
 from typing import Any, Protocol
 
 from context_logger import get_logger
-from zmq import DISH, Poller, POLLIN, POLLOUT, Context
+from zmq import DISH, Poller, POLLIN, Context
 
 from hello import GroupAccess
 
@@ -21,10 +21,13 @@ def start(self, source: GroupAccess) -> None:
     def stop(self) -> None:
         raise NotImplementedError()
 
-    def register(self, callback: OnMessage) -> None:
+    def register(self, handler: OnMessage) -> None:
         raise NotImplementedError()
 
-    def deregister(self, callback: OnMessage) -> None:
+    def deregister(self, handler: OnMessage) -> None:
+        raise NotImplementedError()
+
+    def get_handlers(self) -> list[OnMessage]:
         raise NotImplementedError()
 
 
@@ -37,7 +40,7 @@ def __init__(self, context: Context[Any], max_workers: int = 1, poll_timeout: fl
         self._executor = ThreadPoolExecutor(max_workers=max_workers)
         self._poll_timeout = int(poll_timeout * 1000)
         self._group: str | None = None
-        self._callbacks: list[OnMessage] = []
+        self._handlers: list[OnMessage] = []
 
     def __enter__(self) -> Receiver:
         return self
@@ -53,7 +56,7 @@ def start(self, source: GroupAccess) -> None:
             self._dish.bind(source.access_url)
             self._dish.join(source.full_group)
             self._group = source.full_group
-            self._executor.submit(self._handle_messages)
+            self._executor.submit(self._receive_loop)
             log.debug('Receiver started', address=source.access_url, group=source.full_group)
         except Exception as error:
             log.error('Failed to start receiver', address=source.access_url, group=source.full_group, error=error)
@@ -62,21 +65,23 @@ def start(self, source: GroupAccess) -> None:
     def stop(self) -> None:
         try:
             self._group = None
-            self._poller.register(self._dish, POLLOUT)
             self._executor.shutdown()
             self._dish.close()
             log.debug('Receiver stopped')
         except Exception as error:
             log.error('Failed to stop receiver', error=error)
             raise error
 
-    def register(self, callback: OnMessage) -> None:
-        self._callbacks.append(callback)
+    def register(self, handler: OnMessage) -> None:
+        self._handlers.append(handler)
+
+    def deregister(self, handler: OnMessage) -> None:
+        self._handlers.remove(handler)
 
-    def deregister(self, callback: OnMessage) -> None:
-        self._callbacks.remove(callback)
+    def get_handlers(self) -> list[OnMessage]:
+        return self._handlers.copy()
 
-    def _handle_messages(self) -> None:
+    def _receive_loop(self) -> None:
         while self._group:
             sockets = dict(self._poller.poll(timeout=self._poll_timeout))
             if self._dish in sockets and sockets[self._dish] == POLLIN:
@@ -88,8 +93,8 @@ def _handle_messages(self) -> None:
                     log.error('Failed to receive message', group=self._group, error=error)
 
     def _handle_message(self, message: dict[str, Any]) -> None:
-        for callback in self._callbacks:
+        for handler in self._handlers:
             try:
-                callback(message)
+                handler(message)
             except Exception as error:
-                log.warn('Error in callback execution', data=message, group=self._group, error=error)
+                log.warn('Error in message handler execution', data=message, group=self._group, error=error)

hello/sender.py

@@ -55,8 +55,8 @@ def stop(self) -> None:
 
     def send(self, data: Any) -> None:
         if self._group:
-            if data := self._convert_to_dict(data):
-                self._send_json(data)
+            if message := self._convert_to_dict(data):
+                self._send_json(message)
             else:
                 log.warning('Unsupported message type', data=data, group=self._group)
         else:

tests/advertizerIntegrationTest.py

@@ -14,7 +14,7 @@
 GROUP = Group(GROUP_NAME)
 
 
-class AdvertizerTest(TestCase):
+class AdvertizerIntegrationTest(TestCase):
     SERVICE_INFO = ServiceInfo('test-service', 'test-role', 'http://localhost:8080')
 
     @classmethod
@@ -44,32 +44,6 @@ def test_sends_hello_when_advertises_service(self):
         # Then
         self.assertEqual([self.SERVICE_INFO.__dict__], messages)
 
-    def test_sends_hello_when_advertises_service_and_info_changed(self):
-        # Given
-        context = Context()
-        sender = RadioSender(context)
-        messages = []
-
-        with DefaultAdvertizer(sender) as advertizer, DishReceiver(context) as test_receiver:
-            test_receiver.start(GroupAccess(ACCESS_URL, GROUP.hello()))
-            test_receiver.register(lambda message: messages.append(message))
-            advertizer.start(ACCESS_URL, GROUP)
-
-            advertizer.advertise(self.SERVICE_INFO)
-
-            self.SERVICE_INFO.url = 'http://localhost:9090'
-
-            # When
-            advertizer.advertise(self.SERVICE_INFO)
-
-            wait_for_assertion(0.1, lambda: self.assertEqual(2, len(messages)))
-
-        # Then
-        self.assertEqual([
-            {'name': 'test-service', 'role': 'test-role', 'url': 'http://localhost:8080'},
-            {'name': 'test-service', 'role': 'test-role', 'url': 'http://localhost:9090'}
-        ], messages)
-
     def test_sends_hello_when_query_received(self):
         # Given
         context = Context()

tests/apiIntegrationTest.py

@@ -14,7 +14,7 @@
 SERVICE_QUERY = ServiceQuery('test-service', 'test-role')
 
 
-class ApiTest(TestCase):
+class ApiIntegrationTest(TestCase):
 
     @classmethod
     def setUpClass(cls):
@@ -29,7 +29,7 @@ def test_discoverer_caches_advertised_service(self):
         context = Context()
         hello = DefaultHello(context)
 
-        with hello.default_advertizer() as advertizer, hello.discoverer() as discoverer:
+        with hello.default_advertizer(respond=False) as advertizer, hello.discoverer() as discoverer:
             advertizer.start(ACCESS_URL, GROUP, SERVICE_INFO)
             discoverer.start(ACCESS_URL, GROUP, SERVICE_QUERY)
 
@@ -46,7 +46,7 @@ def test_discoverer_caches_advertised_service_when_scheduled_once(self):
         context = Context()
         hello = DefaultHello(context)
 
-        with hello.scheduled_advertizer() as advertizer, hello.discoverer() as discoverer:
+        with hello.scheduled_advertizer(respond=False) as advertizer, hello.discoverer() as discoverer:
             advertizer.start(ACCESS_URL, GROUP, SERVICE_INFO)
             discoverer.start(ACCESS_URL, GROUP, SERVICE_QUERY)
 
@@ -87,7 +87,7 @@ def test_discoverer_caches_discovery_response_service(self):
             # When
             discoverer.discover()
 
-            wait_for_assertion(0.1, lambda: self.assertEqual(1, len(discoverer.get_services())))
+            wait_for_assertion(0.2, lambda: self.assertEqual(1, len(discoverer.get_services())))
 
         # Then
         self.assertEqual({self.SERVICE_INFO.name: self.SERVICE_INFO}, discoverer.get_services())