Rename event_bus -> bus

Author: emacsway

ascetic_ddd/bus/__init__.py

@@ -0,0 +1,2 @@
+from ascetic_ddd.bus.in_memory_bus import *
+from ascetic_ddd.bus.interfaces import *

…tic_ddd/event_bus/in_memory_event_bus.py ascetic_ddd/bus/in_memory_bus.pyascetic_ddd/event_bus/in_memory_event_bus.py renamed to ascetic_ddd/bus/in_memory_bus.py ascetic_ddd/event_bus/in_memory_event_bus.py renamed to ascetic_ddd/bus/in_memory_bus.py

@@ -1,30 +1,30 @@
 import collections
 import typing
 
+from ascetic_ddd.bus.interfaces import IBus, IHandler
 from ascetic_ddd.disposable import Disposable, IDisposable
-from ascetic_ddd.event_bus.interfaces import IEventBus, IEventHandler
 
-__all__ = ("InMemoryEventBus",)
+__all__ = ("InMemoryBus",)
 
 SessionT = typing.TypeVar("SessionT")
-EventT = typing.TypeVar("EventT")
+MessageT = typing.TypeVar("MessageT")
 
 
-class InMemoryEventBus(IEventBus[SessionT], typing.Generic[SessionT]):
+class InMemoryBus(IBus[SessionT], typing.Generic[SessionT]):
     def __init__(self) -> None:
         self._subscribers: collections.defaultdict[str, list] = collections.defaultdict(list)
 
-    async def publish(self, session: SessionT, uri: str, event: EventT) -> None:
+    async def publish(self, session: SessionT, uri: str, message: MessageT) -> None:
         for handler in self._subscribers[uri]:
-            await handler(session, uri, event)
+            await handler(session, uri, message)
 
-    async def subscribe(self, uri: str, handler: IEventHandler[SessionT, EventT]) -> IDisposable:
+    async def subscribe(self, uri: str, handler: IHandler[SessionT, MessageT]) -> IDisposable:
         self._subscribers[uri].append(handler)
 
         async def callback() -> None:
             await self.unsubscribe(uri, handler)
 
         return Disposable(callback)
 
-    async def unsubscribe(self, uri: str, handler: IEventHandler[SessionT, EventT]) -> None:
+    async def unsubscribe(self, uri: str, handler: IHandler[SessionT, MessageT]) -> None:
         self._subscribers[uri].remove(handler)

ascetic_ddd/bus/interfaces.py

@@ -0,0 +1,34 @@
+import typing
+from abc import ABCMeta, abstractmethod
+
+from ascetic_ddd.disposable import IDisposable
+
+__all__ = (
+    "IBus",
+    "IHandler",
+)
+
+SessionT = typing.TypeVar("SessionT")
+MessageT = typing.TypeVar("MessageT")
+
+SessionT_contra = typing.TypeVar("SessionT_contra", contravariant=True)
+MessageT_contra = typing.TypeVar("MessageT_contra", contravariant=True)
+
+
+class IHandler(typing.Protocol[SessionT_contra, MessageT_contra]):
+    def __call__(self, session: SessionT_contra, uri: str, message: MessageT_contra):
+        ...
+
+
+class IBus(typing.Generic[SessionT], metaclass=ABCMeta):
+    @abstractmethod
+    async def publish(self, session: SessionT, uri: str, message: MessageT) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    async def subscribe(self, uri: str, handler: IHandler[SessionT, MessageT]) -> IDisposable:
+        raise NotImplementedError
+
+    @abstractmethod
+    async def unsubscribe(self, uri: str, handler: IHandler[SessionT, MessageT]) -> None:
+        raise NotImplementedError

ascetic_ddd/event_bus/tests/__init__.py ascetic_ddd/bus/tests/__init__.pyascetic_ddd/event_bus/tests/__init__.py renamed to ascetic_ddd/bus/tests/__init__.py

@@ -3,45 +3,45 @@
 from contextlib import contextmanager
 from unittest import IsolatedAsyncioTestCase, mock
 
-from ascetic_ddd.event_bus.in_memory_event_bus import InMemoryEventBus
+from ascetic_ddd.bus.in_memory_bus import InMemoryBus
 
 
-class InMemoryEventBusTestCase(IsolatedAsyncioTestCase):
+class InMemoryBusTestCase(IsolatedAsyncioTestCase):
     session = object()
     uri = "sb://test-uri"
-    event = {"a": 5}
+    message = {"a": 5}
 
     def setUp(self) -> None:
-        self.event_bus = InMemoryEventBus()
+        self.bus = InMemoryBus()
 
     async def test_publish(self):
         handler = mock.AsyncMock()
-        await self.event_bus.subscribe(self.uri, handler)
+        await self.bus.subscribe(self.uri, handler)
         with self._wait_for_response_until(lambda: handler.called):
-            await self.event_bus.publish(self.session, self.uri, self.event)
-        handler.assert_called_once_with(self.session, self.uri, self.event)
+            await self.bus.publish(self.session, self.uri, self.message)
+        handler.assert_called_once_with(self.session, self.uri, self.message)
 
     async def test_unsubscribe(self):
         handler = mock.AsyncMock()
         handler2 = mock.AsyncMock()
-        await self.event_bus.subscribe(self.uri, handler)
-        await self.event_bus.subscribe(self.uri, handler2)
-        await self.event_bus.unsubscribe(self.uri, handler)
+        await self.bus.subscribe(self.uri, handler)
+        await self.bus.subscribe(self.uri, handler2)
+        await self.bus.unsubscribe(self.uri, handler)
         with self._wait_for_response_until(lambda: handler2.called):
-            await self.event_bus.publish(self.session, self.uri, self.event)
+            await self.bus.publish(self.session, self.uri, self.message)
         handler.assert_not_called()
-        handler2.assert_called_once_with(self.session, self.uri, self.event)
+        handler2.assert_called_once_with(self.session, self.uri, self.message)
 
-    async def test_disposable_event(self):
+    async def test_disposable(self):
         handler = mock.AsyncMock()
         handler2 = mock.AsyncMock()
-        disposable = await self.event_bus.subscribe(self.uri, handler)
-        await self.event_bus.subscribe(self.uri, handler2)
+        disposable = await self.bus.subscribe(self.uri, handler)
+        await self.bus.subscribe(self.uri, handler2)
         await disposable.dispose()
         with self._wait_for_response_until(lambda: handler2.called):
-            await self.event_bus.publish(self.session, self.uri, self.event)
+            await self.bus.publish(self.session, self.uri, self.message)
         handler.assert_not_called()
-        handler2.assert_called_once_with(self.session, self.uri, self.event)
+        handler2.assert_called_once_with(self.session, self.uri, self.message)
 
     @staticmethod
     def _wait_for_response_until(*checkers):

…dd/event_bus/tests/event_bus/__init__.py ascetic_ddd/bus/tests/bus/__init__.pyascetic_ddd/event_bus/tests/event_bus/__init__.py renamed to ascetic_ddd/bus/tests/bus/__init__.py ascetic_ddd/event_bus/tests/event_bus/__init__.py renamed to ascetic_ddd/bus/tests/bus/__init__.py

@@ -1,4 +1,4 @@
-from ascetic_ddd.event_bus import IEventBus, InMemoryEventBus
+from ascetic_ddd.bus import IBus, InMemoryBus
 from ascetic_ddd.session.interfaces import ISession
 from ascetic_ddd.utils.amemo import amemo
 
@@ -10,8 +10,8 @@
 
 class BuildingBlocksFactory:
     @amemo
-    async def make_in_memory_event_bus(self) -> IEventBus[ISession]:
-        return InMemoryEventBus[ISession]()
+    async def make_in_memory_bus(self) -> IBus[ISession]:
+        return InMemoryBus[ISession]()
 
 
 ascetic_ddd_factory = BuildingBlocksFactory()

…nt_bus/tests/event_bus/test_event_bus.py ascetic_ddd/bus/tests/bus/test_bus.pyascetic_ddd/event_bus/tests/event_bus/test_event_bus.py renamed to ascetic_ddd/bus/tests/bus/test_bus.py ascetic_ddd/event_bus/tests/event_bus/test_event_bus.py renamed to ascetic_ddd/bus/tests/bus/test_bus.py

@@ -41,7 +41,7 @@ Package Structure
    +-- outbox/            # Transactional outbox
    +-- inbox/             # Idempotent inbox
    +-- saga/              # Saga orchestration
-   +-- event_bus/         # In-process event bus
+   +-- bus/               # In-process message bus
    +-- mediator/          # Command/query mediator
    +-- signals/           # Typed Signal pattern
    +-- session/           # Session/UoW management