Deprecate channels
@@ -5,8 +5,11 @@ from inspect import isawaitable, iscoroutinefunction
from typing import Any, Type, Callable, Tuple, Optional, Dict
-# TODO: After playing with concept a bit it feels like we don't actually need this
-# abstraction, and getting rid of it can make code more straightforward
+# TODO: DEPRECATE
+# After playing with concept a bit it feels like using EventBus
+# is more appropriate.
+
+
class TxChannel(ABC):
@abstractmethod
async def put(self, msg: Any):
@@ -17,19 +20,12 @@ class TxChannel(ABC):
pass
-# TODO: Rename to stream/observable?
class RxChannel(ABC):
@abstractmethod
- # TODO: Optimization - adding subscriptions per type can reduce need for
- # separate router and improve performance. However coupling implications
- # need more analysis.
def subscribe(self, handler: Callable):
pass
-# TODO: We don't have async handlers now - so we don't need this complexity
-# If we ever actually need async handlers then we might just as well create
-# separate AsyncRxChannel/AsyncStream
class _RxChannel(RxChannel):
def __init__(self):
self._sync_handlers = []
@@ -55,8 +51,6 @@ class _RxChannel(RxChannel):
await wait([handler(msg) for handler in self._async_handlers])
-# TODO: Having single DirectChannel implementing RxChannel interface would make
-# usage much simpler
def create_direct_channel() -> Tuple[TxChannel, RxChannel]:
class Channel(TxChannel):
def __init__(self, observable: _RxChannel):
@@ -74,8 +68,6 @@ def create_direct_channel() -> Tuple[TxChannel, RxChannel]:
return Channel(router), router
-# TODO: This was a PoC implementation to show how current queues could be implemented
-# using channel framework. This is actually not used and untested, probably remove it?
class QueuedChannelService:
class Channel(TxChannel):
def __init__(self, queue: deque):
@@ -115,9 +107,6 @@ class QueuedChannelService:
return count
-# TODO: This was a PoC implementation to show how transparent port to AsyncIO could be
-# performed using channel framework. This is actually not used and untested, probably
-# remove it?
class AsyncioChannelService:
class Channel(TxChannel):
def __init__(self, queue: Queue):
@@ -155,9 +144,6 @@ class AsyncioChannelService:
self._is_running = False
-# TODO: If we make default channel synchronous then we can just merge this
-# with Router. If we ever need async router we can just create a new implementation
-# relying on async channels and coroutines
class RouterBase:
def __init__(self, strict: bool = False):
self._strict = strict
@@ -196,7 +182,6 @@ class Router(RouterBase):
RouterBase.add(self, msg_type, handler)
-# TODO: We don't have async handlers now, remove it?
class AsyncRouter(RouterBase):
def __init__(self, input: RxChannel, strict: bool = True):
RouterBase.__init__(self, strict)
