#70 Introducing a host of listeners in the bulkhead component (#73)

* #70 Introducing a host of listeners in the bulkhead component

* #70 Covered bulkhead listeners by tests

Author: roma-glushko

hyx/bulkhead/__init__.py

@@ -1,3 +1,4 @@
 from hyx.bulkhead.api import bulkhead
+from hyx.bulkhead.listeners import BulkheadListener, register_bulkhead_listener
 
-__all__ = ("bulkhead",)
+__all__ = ("bulkhead", "BulkheadListener", "register_bulkhead_listener")

hyx/bulkhead/api.py

@@ -1,8 +1,10 @@
 import functools
 from types import TracebackType
-from typing import Any, Optional, Type, cast
+from typing import Any, Optional, Sequence, Type, cast
 
+from hyx.bulkhead.listeners import BulkheadListener
 from hyx.bulkhead.manager import BulkheadManager
+from hyx.common.events import EventDispatcher
 from hyx.common.typing import FuncT
 
 
@@ -22,10 +24,20 @@ class bulkhead:
 
     __slots__ = ("_manager",)
 
-    def __init__(self, max_concurrency: int, max_capacity: int) -> None:
+    def __init__(
+        self,
+        max_concurrency: int,
+        max_capacity: int,
+        *,
+        name: Optional[str] = None,
+        listeners: Optional[Sequence[BulkheadListener]] = None,
+    ) -> None:
+
         self._manager = BulkheadManager(
             max_concurrency=max_concurrency,
             max_capacity=max_capacity,
+            name=name,
+            event_dispatcher=EventDispatcher(listeners).as_listener,
         )
 
     async def __aenter__(self) -> "bulkhead":

hyx/bulkhead/listeners.py

@@ -0,0 +1,22 @@
+from typing import TYPE_CHECKING
+
+from hyx.common.events import ListenerRegistry
+
+if TYPE_CHECKING:
+    from hyx.bulkhead.manager import BulkheadManager
+
+_BULKHEAD_LISTENERS: ListenerRegistry["BulkheadListener"] = ListenerRegistry()
+
+
+class BulkheadListener:
+    async def on_bulkhead_full(self, bulkhead: "BulkheadManager") -> None:
+        ...
+
+
+def register_bulkhead_listener(listener: BulkheadListener) -> None:
+    """
+    Register a listener that will listen to all fallback components in the system
+    """
+    global _BULKHEAD_LISTENERS
+
+    _BULKHEAD_LISTENERS.register(listener)

hyx/bulkhead/manager.py

@@ -1,7 +1,8 @@
 import asyncio
-from typing import Any
+from typing import Any, Optional
 
 from hyx.bulkhead.exceptions import BulkheadFull
+from hyx.bulkhead.listeners import BulkheadListener
 from hyx.common.typing import FuncT
 
 
@@ -10,9 +11,15 @@ class BulkheadManager:
     Semaphore-based bulkhead implementation
     """
 
-    __slots__ = ("_total_execs_limiter", "_concurrency_limiter")
+    __slots__ = ("_total_execs_limiter", "_concurrency_limiter", "_name", "_event_dispatcher")
 
-    def __init__(self, max_concurrency: int, max_capacity: int) -> None:
+    def __init__(
+        self,
+        max_concurrency: int,
+        max_capacity: int,
+        event_dispatcher: BulkheadListener,
+        name: Optional[str] = None,
+    ) -> None:
         if max_concurrency <= 0:
             raise ValueError(f'max_concurrency should be greater than zero ("{max_concurrency}" given)')
 
@@ -25,12 +32,17 @@ def __init__(self, max_concurrency: int, max_capacity: int) -> None:
         self._concurrency_limiter = asyncio.Semaphore(max_concurrency)
         self._total_execs_limiter = asyncio.Semaphore(max_capacity)
 
-    def _raise_on_exceed(self) -> None:
+        self._name = name
+        self._event_dispatcher = event_dispatcher
+
+    async def _raise_on_exceed(self) -> None:
         if self._total_execs_limiter.locked():
+            await self._event_dispatcher.on_bulkhead_full(self)
+
             raise BulkheadFull
 
     async def acquire(self) -> None:
-        self._raise_on_exceed()
+        await self._raise_on_exceed()
 
         await self._total_execs_limiter.acquire()
         await self._concurrency_limiter.acquire()
@@ -40,7 +52,7 @@ async def release(self) -> None:
         self._total_execs_limiter.release()
 
     async def __call__(self, func: FuncT) -> Any:
-        self._raise_on_exceed()
+        await self._raise_on_exceed()
 
         async with self._total_execs_limiter:
             async with self._concurrency_limiter:

tests/test_bulkhead/test_api.py

@@ -1,19 +1,35 @@
 import asyncio
+from unittest.mock import Mock
 
 import pytest
 
-from hyx.bulkhead import bulkhead
+from hyx.bulkhead import BulkheadListener, bulkhead
 from hyx.bulkhead.exceptions import BulkheadFull
+from hyx.bulkhead.manager import BulkheadManager
+from tests.conftest import event_manager
+
+
+class Listener(BulkheadListener):
+    def __init__(self) -> None:
+        self.is_full = Mock()
+
+    async def on_bulkhead_full(self, bulkhead: "BulkheadManager") -> None:
+        self.is_full()
 
 
 async def test__bulkhead__decorator() -> None:
-    @bulkhead(max_capacity=3, max_concurrency=2)
+    listener = Listener()
+
+    @bulkhead(max_capacity=3, max_concurrency=2, listeners=(listener,))
     async def calculations() -> float:
         await asyncio.sleep(0.5)
         return 42
 
     assert await calculations() == 42
 
+    await event_manager.wait_for_tasks()
+    listener.is_full.assert_not_called()
+
 
 async def test__bulkhead__context() -> None:
     async with bulkhead(max_capacity=3, max_concurrency=2):
@@ -22,14 +38,18 @@ async def test__bulkhead__context() -> None:
 
 
 async def test__bulkhead__capacity_exceeded() -> None:
-    bh = bulkhead(max_capacity=2, max_concurrency=2)
+    listener = Listener()
+    bh = bulkhead(max_capacity=2, max_concurrency=2, listeners=(listener,))
 
     with pytest.raises(BulkheadFull):
         async with bh:
             async with bh:
                 async with bh:
                     assert True
 
+    await event_manager.wait_for_tasks()
+    listener.is_full.assert_called()
+
 
 async def test__bulkhead__capacity_exceeded_from_different_coroutines() -> None:
     bh = bulkhead(max_capacity=3, max_concurrency=2)