Initial commit WIP

Author: AttilaGombosER

hello/advertizer.py

@@ -1,3 +1,5 @@
+import random
+import time
 from typing import Any
 
 from common_utility import IReusableTimer
@@ -47,46 +49,49 @@ def advertise(self, info: ServiceInfo | None = None) -> None:
             if info:
                 self._info = info
             if self._info:
-                self._sender.send(info)
+                self._sender.send(self._info)
                 log.info('Service advertised', service=self._info, group=self._group)
         else:
             log.warning('Cannot advertise service, advertizer not started', service=info)
 
 
 class RespondingAdvertizer(DefaultAdvertizer):
 
-    def __init__(self, sender: Sender, receiver: Receiver) -> None:
+    def __init__(self, sender: Sender, receiver: Receiver, max_response_delay: float = 0.1) -> None:
         super().__init__(sender)
         self._receiver = receiver
+        self._max_delay = max_response_delay
 
     def start(self, address: str, group: Group, info: ServiceInfo | None = None) -> None:
         super().start(address, group, info)
         self._receiver.start(GroupAccess(address, group.query()))
-        self._receiver.register(self._handle_query)
+        self._receiver.register(self._handle_message)
 
     def stop(self) -> None:
         super().stop()
         self._receiver.stop()
 
-    def _handle_query(self, data: dict[str, str]) -> None:
+    def _handle_message(self, message: dict[str, Any]) -> None:
         if self._info:
-            matcher: ServiceMatcher | None = None
-
             try:
-                query = ServiceQuery(**data)
-                matcher = ServiceMatcher(query)
+                query = ServiceQuery(**message)
                 log.debug('Query received', group=self._group, query=query)
+                self._handle_query(query, self._info)
             except Exception as error:
-                log.warning('Invalid query message received', group=self._group, received=data, error=error)
+                log.warning('Invalid query message received', group=self._group, received=message, error=error)
 
-            if matcher and matcher.matches(self._info):
-                log.info('Query matches service', group=self._group, query=matcher.query, service=self._info)
-                self.advertise(self._info)
+    def _handle_query(self, query: ServiceQuery, info: ServiceInfo) -> None:
+        matcher = ServiceMatcher(query)
+        if matcher and matcher.matches(info):
+            delay = round(self._max_delay * random.random(), 3)
+            log.info('Responding to query', group=self._group, query=matcher.query, service=info, delay=delay)
+            time.sleep(delay)
+            self.advertise(info)
 
 
 class ScheduledAdvertizer(Advertizer):
 
-    def schedule(self, info: ServiceInfo, interval: float, one_shot: bool = False) -> None:
+    def schedule(self, info: ServiceInfo | None = None, interval: float = 10, one_shot: bool = False) -> None:
         raise NotImplementedError()
 
 
@@ -112,7 +117,7 @@ def stop(self) -> None:
     def advertise(self, info: ServiceInfo | None = None) -> None:
         self._advertizer.advertise(info)
 
-    def schedule(self, info: ServiceInfo, interval: float, one_shot: bool = False) -> None:
+    def schedule(self, info: ServiceInfo | None = None, interval: float = 10, 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)

hello/api.py

@@ -1,6 +1,6 @@
 from typing import Any
 
-from common_utility import ReusableTimer, IReusableTimer
+from common_utility import ReusableTimer
 from zmq import Context
 
 from hello import Advertizer, Discoverer, RadioSender, DishReceiver, DefaultAdvertizer, DefaultDiscoverer, \
@@ -9,10 +9,10 @@
 
 class Hello:
 
-    def default_advertizer(self, respond: bool = True) -> Advertizer:
+    def default_advertizer(self, respond: bool = True, delay: float = 0.1) -> Advertizer:
         raise NotImplementedError()
 
-    def scheduled_advertizer(self, timer: IReusableTimer | None = None, respond: bool = True) -> ScheduledAdvertizer:
+    def scheduled_advertizer(self, respond: bool = True, delay: float = 0.1) -> ScheduledAdvertizer:
         raise NotImplementedError()
 
     def discoverer(self) -> Discoverer:
@@ -21,18 +21,24 @@ def discoverer(self) -> Discoverer:
 
 class DefaultHello(Hello):
 
-    def __init__(self, context: Context[Any] | None = None) -> None:
+    def __init__(self, context: Context[Any] | None = None, max_workers: int = 1, poll_timeout: float = 0.1) -> None:
         self._context = context if context else Context()
-        self._sender = RadioSender(self._context)
-        self._receiver = DishReceiver(self._context)
+        self._max_workers = max_workers
+        self._poll_timeout = poll_timeout
 
-    def default_advertizer(self, respond: bool = True) -> Advertizer:
-        return RespondingAdvertizer(self._sender, self._receiver) if respond else DefaultAdvertizer(self._sender)
+    def default_advertizer(self, respond: bool = True, delay: float = 0.1) -> Advertizer:
+        sender = RadioSender(self._context)
+        if respond:
+            receiver = DishReceiver(self._context, self._max_workers, self._poll_timeout)
+            return RespondingAdvertizer(sender, receiver, delay)
+        else:
+            return DefaultAdvertizer(sender)
 
-    def scheduled_advertizer(self, timer: IReusableTimer | None = None, respond: bool = True) -> ScheduledAdvertizer:
-        advertizer = self.default_advertizer(respond)
-        reusable_timer = timer if timer else ReusableTimer()
-        return DefaultScheduledAdvertizer(advertizer, reusable_timer)
+    def scheduled_advertizer(self, respond: bool = True, delay: float = 0.1) -> ScheduledAdvertizer:
+        advertizer = self.default_advertizer(respond, delay)
+        return DefaultScheduledAdvertizer(advertizer, ReusableTimer())
 
     def discoverer(self) -> Discoverer:
-        return DefaultDiscoverer(self._sender, self._receiver)
+        sender = RadioSender(self._context)
+        receiver = DishReceiver(self._context, self._max_workers, self._poll_timeout)
+        return DefaultDiscoverer(sender, receiver)

hello/discoverer.py

@@ -1,4 +1,6 @@
-from typing import Any, Callable
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any, Protocol
 
 from context_logger import get_logger
 
@@ -7,6 +9,21 @@
 log = get_logger('Discoverer')
 
 
+class DiscoveryEventType(Enum):
+    DISCOVERED = 'discovered'
+    UPDATED = 'updated'
+
+
+@dataclass
+class DiscoveryEvent:
+    service: ServiceInfo
+    type: DiscoveryEventType
+
+
+class OnDiscoveryEvent(Protocol):
+    def __call__(self, event: DiscoveryEvent) -> None: ...
+
+
 class Discoverer:
 
     def start(self, address: str, group: Group, query: ServiceQuery | None = None) -> None:
@@ -21,10 +38,10 @@ def discover(self, query: ServiceQuery | None = None) -> None:
     def get_services(self) -> dict[str, ServiceInfo]:
         raise NotImplementedError()
 
-    def register(self, callback: Callable[[Any], None]) -> None:
+    def register(self, callback: OnDiscoveryEvent) -> None:
         raise NotImplementedError()
 
-    def deregister(self, callback: Callable[[Any], None]) -> None:
+    def deregister(self, callback: OnDiscoveryEvent) -> None:
         raise NotImplementedError()
 
 
@@ -36,7 +53,7 @@ def __init__(self, sender: Sender, receiver: Receiver) -> None:
         self._group: Group | None = None
         self._matcher: ServiceMatcher | None = None
         self._services: dict[str, ServiceInfo] = {}
-        self._callbacks: list[Callable[[ServiceInfo], None]] = []
+        self._callbacks: list[OnDiscoveryEvent] = []
 
     def __enter__(self) -> Discoverer:
         return self
@@ -65,24 +82,24 @@ def discover(self, query: ServiceQuery | None = None) -> None:
                 self._sender.send(self._matcher.query)
                 log.info('Service discovery initiated', query=self._matcher.query, group=self._group)
         else:
-            log.warning('Cannot initiate service discovery, discoverer not started', query=query)
+            log.warning('Cannot discover services, discoverer not started', query=query)
 
     def get_services(self) -> dict[str, ServiceInfo]:
         return self._services.copy()
 
-    def register(self, callback: Callable[[Any], None]) -> None:
+    def register(self, callback: OnDiscoveryEvent) -> None:
         self._callbacks.append(callback)
 
-    def deregister(self, callback: Callable[[Any], None]) -> None:
+    def deregister(self, callback: OnDiscoveryEvent) -> None:
         self._callbacks.remove(callback)
 
-    def _handle_message(self, data: dict[str, Any]) -> None:
+    def _handle_message(self, message: dict[str, Any]) -> None:
         service: ServiceInfo | None = None
 
         try:
-            service = ServiceInfo(**data)
+            service = ServiceInfo(**message)
         except Exception as error:
-            log.warn('Failed to handle received message', data=data, error=error)
+            log.warn('Failed to handle received message', data=message, error=error)
 
         if service:
             self._handle_service(service)
@@ -91,29 +108,25 @@ def _handle_service(self, service: ServiceInfo) -> None:
         if self._matcher and self._matcher.matches(service):
             cached = self._services.get(service.name)
 
-            if self._is_update_needed(cached, service):
-                self._services[service.name] = service
-                for callback in self._callbacks:
-                    try:
-                        callback(service)
-                    except Exception as error:
-                        log.warn('Error in callback execution', service=service, error=error)
+            if event := self._create_event(cached, service):
+                self._handle_event(event)
 
-    def _is_update_needed(self, cached: ServiceInfo | None, service: ServiceInfo) -> bool:
+    def _create_event(self, cached: ServiceInfo | None, service: ServiceInfo) -> DiscoveryEvent | None:
         if cached:
             if cached != service:
                 log.info('Service updated', old_service=cached, new_service=service)
-                return True
+                return DiscoveryEvent(service, DiscoveryEventType.UPDATED)
         else:
             log.info('Service discovered', service=service)
-            return True
+            return DiscoveryEvent(service, DiscoveryEventType.DISCOVERED)
 
-        return False
+        return None
 
-    def _handle_update(self, service: ServiceInfo) -> None:
+    def _handle_event(self, event: DiscoveryEvent) -> None:
+        service = event.service
         self._services[service.name] = service
         for callback in self._callbacks:
             try:
-                callback(service)
+                callback(event)
             except Exception as error:
                 log.warn('Error in callback execution', service=service, error=error)

hello/receiver.py

@@ -1,5 +1,5 @@
 from concurrent.futures import ThreadPoolExecutor
-from typing import Any, Callable
+from typing import Any, Protocol
 
 from context_logger import get_logger
 from zmq import DISH, Poller, POLLIN, POLLOUT, Context
@@ -9,6 +9,10 @@
 log = get_logger('Receiver')
 
 
+class OnMessage(Protocol):
+    def __call__(self, message: dict[str, Any]) -> None: ...
+
+
 class Receiver:
 
     def start(self, source: GroupAccess) -> None:
@@ -17,22 +21,23 @@ def start(self, source: GroupAccess) -> None:
     def stop(self) -> None:
         raise NotImplementedError()
 
-    def register(self, callback: Callable[[Any], None]) -> None:
+    def register(self, callback: OnMessage) -> None:
         raise NotImplementedError()
 
-    def deregister(self, callback: Callable[[Any], None]) -> None:
+    def deregister(self, callback: OnMessage) -> None:
         raise NotImplementedError()
 
 
 class DishReceiver(Receiver):
 
-    def __init__(self, context: Context[Any]) -> None:
+    def __init__(self, context: Context[Any], max_workers: int = 1, poll_timeout: float = 0.1) -> None:
         self._context = context
         self._dish = self._context.socket(DISH)
         self._poller = Poller()
-        self._executor = ThreadPoolExecutor(max_workers=1)
+        self._executor = ThreadPoolExecutor(max_workers=max_workers)
+        self._poll_timeout = int(poll_timeout * 1000)
         self._group: str | None = None
-        self._callbacks: list[Callable[[Any], None]] = []
+        self._callbacks: list[OnMessage] = []
 
     def __enter__(self) -> Receiver:
         return self
@@ -42,12 +47,14 @@ def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
 
     def start(self, source: GroupAccess) -> None:
         try:
+            if self._group:
+                raise RuntimeError('Receiver already started')
             self._poller.register(self._dish, POLLIN)
             self._dish.bind(source.access_url)
             self._dish.join(source.full_group)
             self._group = source.full_group
             self._executor.submit(self._handle_messages)
-            log.info('Receiver started', address=source.access_url, group=source.full_group)
+            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)
             raise error
@@ -58,20 +65,20 @@ def stop(self) -> None:
             self._poller.register(self._dish, POLLOUT)
             self._executor.shutdown()
             self._dish.close()
-            log.info('Receiver stopped')
+            log.debug('Receiver stopped')
         except Exception as error:
             log.error('Failed to stop receiver', error=error)
             raise error
 
-    def register(self, callback: Callable[[Any], None]) -> None:
+    def register(self, callback: OnMessage) -> None:
         self._callbacks.append(callback)
 
-    def deregister(self, callback: Callable[[Any], None]) -> None:
+    def deregister(self, callback: OnMessage) -> None:
         self._callbacks.remove(callback)
 
     def _handle_messages(self) -> None:
         while self._group:
-            sockets = dict(self._poller.poll(timeout=100))
+            sockets = dict(self._poller.poll(timeout=self._poll_timeout))
             if self._dish in sockets and sockets[self._dish] == POLLIN:
                 try:
                     data = self._dish.recv_json()
@@ -80,9 +87,9 @@ def _handle_messages(self) -> None:
                 except Exception as error:
                     log.error('Failed to receive message', group=self._group, error=error)
 
-    def _handle_message(self, data: dict[str, str]) -> None:
+    def _handle_message(self, message: dict[str, Any]) -> None:
         for callback in self._callbacks:
             try:
-                callback(data)
+                callback(message)
             except Exception as error:
-                log.warn('Error in callback execution', data=data, group=self._group, error=error)
+                log.warn('Error in callback execution', data=message, group=self._group, error=error)

hello/sender.py

@@ -35,9 +35,11 @@ def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
 
     def start(self, target: GroupAccess) -> None:
         try:
+            if self._group:
+                raise RuntimeError('Sender already started')
             self._radio.connect(target.access_url)
             self._group = target.full_group
-            log.info('Sender started', address=target.access_url, group=target.full_group)
+            log.debug('Sender started', address=target.access_url, group=target.full_group)
         except Exception as error:
             log.error('Failed to start sender', address=target.access_url, group=target.full_group, error=error)
             raise error
@@ -46,7 +48,7 @@ def stop(self) -> None:
         try:
             self._group = None
             self._radio.close()
-            log.info('Sender stopped')
+            log.debug('Sender stopped')
         except Exception as error:
             log.error('Failed to stop sender', error=error)
             raise error
@@ -57,6 +59,8 @@ def send(self, data: Any) -> None:
                 self._send_json(data)
             else:
                 log.warning('Unsupported message type', data=data, group=self._group)
+        else:
+            log.warning('Cannot send message, sender not started', data=data)
 
     def _convert_to_dict(self, data: Any) -> dict[str, Any] | None:
         if isinstance(data, dict):