refactor(channels): remove dead MessageBus dispatcher (#205)
* refactor(channels): remove dead MessageBus dispatcher Outbound routing has two implementations: ``MessageBus.dispatch_outbound`` (subscriber-based) and ``ChannelManager._dispatch_outbound`` (registry lookup). Only the latter is ever started in production — the former is reachable solely from tests, yet both consume from the same ``bus.outbound`` queue. If anyone followed the bus's own API surface they would silently steal messages from the real dispatcher. Drop the unused machinery to leave a single, obvious outbound path: - ``MessageBus.subscribe_outbound`` / ``dispatch_outbound`` / ``stop`` - ``_running`` flag and ``_outbound_subscribers`` map - ``OutboundCallback`` type alias - The lone ``bus.stop()`` call in ``cli/channel.py`` (was no-op) - Four tests covering the removed code paths * test(channels): drop empty MessageBus stubs after dispatcher removal --------- Co-authored-by: Xi Zhang <106144707+X-iZhang@users.noreply.github.com>
This commit is contained in:
Ziheng Zhang
@@ -2,23 +2,20 @@
|
||||
|
||||
Channels push messages to the inbound queue; the agent (or any consumer)
|
||||
reads from inbound, processes, and pushes responses to the outbound queue.
|
||||
A background dispatcher routes outbound messages to the correct channel
|
||||
via subscriber callbacks.
|
||||
``ChannelManager._dispatch_outbound`` routes outbound messages to the
|
||||
correct channel by looking up its registered :class:`Channel` instance.
|
||||
|
||||
Deduplication is handled at the Channel level (single dedup point).
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from collections.abc import Awaitable, Callable
|
||||
|
||||
from ..debug import TraceMixin, debug_trace_enabled
|
||||
from .events import InboundMessage, OutboundMessage
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
OutboundCallback = Callable[[OutboundMessage], Awaitable[None]]
|
||||
|
||||
|
||||
class MessageBus(TraceMixin):
|
||||
"""Async message bus that decouples chat channels from the agent core."""
|
||||
@@ -28,8 +25,6 @@ class MessageBus(TraceMixin):
|
||||
def __init__(self):
|
||||
self.inbound: asyncio.Queue[InboundMessage] = asyncio.Queue(maxsize=5000)
|
||||
self.outbound: asyncio.Queue[OutboundMessage] = asyncio.Queue(maxsize=5000)
|
||||
self._outbound_subscribers: dict[str, list[OutboundCallback]] = {}
|
||||
self._running = False
|
||||
self._debug_trace = debug_trace_enabled()
|
||||
self._trace_logger = logger
|
||||
|
||||
@@ -53,58 +48,6 @@ class MessageBus(TraceMixin):
|
||||
"""Consume the next outbound message (blocks until available)."""
|
||||
return await self.outbound.get()
|
||||
|
||||
# ── subscriber routing ──
|
||||
|
||||
def subscribe_outbound(
|
||||
self,
|
||||
channel: str,
|
||||
callback: OutboundCallback,
|
||||
) -> None:
|
||||
"""Register a callback for outbound messages targeting *channel*."""
|
||||
if channel not in self._outbound_subscribers:
|
||||
self._outbound_subscribers[channel] = []
|
||||
self._outbound_subscribers[channel].append(callback)
|
||||
|
||||
async def dispatch_outbound(self) -> None:
|
||||
"""Route outbound messages to subscribed channels.
|
||||
|
||||
Run as a background task — loops until :meth:`stop` is called.
|
||||
"""
|
||||
self._running = True
|
||||
while self._running:
|
||||
try:
|
||||
msg = await asyncio.wait_for(
|
||||
self.outbound.get(),
|
||||
timeout=1.0,
|
||||
)
|
||||
except TimeoutError:
|
||||
continue
|
||||
subscribers = self._outbound_subscribers.get(msg.channel, [])
|
||||
if not subscribers:
|
||||
self._trace_event(
|
||||
"bus_dispatch_drop",
|
||||
target_channel=msg.channel,
|
||||
reason="no_subscriber",
|
||||
chat_id=msg.chat_id,
|
||||
)
|
||||
logger.warning(f"No subscriber for channel: {msg.channel}")
|
||||
continue
|
||||
for callback in subscribers:
|
||||
try:
|
||||
await callback(msg)
|
||||
except Exception as e:
|
||||
self._trace_event(
|
||||
"bus_dispatch_error",
|
||||
target_channel=msg.channel,
|
||||
chat_id=msg.chat_id,
|
||||
error_type=type(e).__name__,
|
||||
)
|
||||
logger.error(f"Error dispatching to {msg.channel}: {e}")
|
||||
|
||||
def stop(self) -> None:
|
||||
"""Stop the dispatcher loop."""
|
||||
self._running = False
|
||||
|
||||
@property
|
||||
def inbound_size(self) -> int:
|
||||
return self.inbound.qsize()
|
||||
|
||||
@@ -741,8 +741,6 @@ def _channels_stop(
|
||||
future.result(timeout=10)
|
||||
except Exception as e:
|
||||
_channel_logger.debug(f"Error stopping channels: {e}")
|
||||
if _manager:
|
||||
_manager.bus.stop()
|
||||
if _bus_thread:
|
||||
_bus_thread.join(timeout=5)
|
||||
_manager = None
|
||||
|
||||
@@ -1510,6 +1510,13 @@ class TestInboundConsumerErrorHandling:
|
||||
|
||||
|
||||
class TestMessageBus:
|
||||
"""Covers the bus as a pure pub/sub queue.
|
||||
|
||||
Outbound routing (subscriber dispatch, error handling, stop semantics)
|
||||
is owned by ``ChannelManager._dispatch_outbound`` — see
|
||||
``TestChannelManagerDispatch`` for that coverage.
|
||||
"""
|
||||
|
||||
def test_publish_consume_inbound(self):
|
||||
async def _test():
|
||||
bus = MessageBus()
|
||||
@@ -1535,57 +1542,6 @@ class TestMessageBus:
|
||||
|
||||
_run(_test())
|
||||
|
||||
def test_subscriber_dispatch(self):
|
||||
async def _test():
|
||||
bus = MessageBus()
|
||||
received = []
|
||||
bus.subscribe_outbound("tg", lambda m: received.append(m))
|
||||
|
||||
task = asyncio.create_task(bus.dispatch_outbound())
|
||||
await bus.publish_outbound(
|
||||
BusOutbound(
|
||||
channel="tg",
|
||||
chat_id="c1",
|
||||
content="hello",
|
||||
)
|
||||
)
|
||||
await asyncio.sleep(0.1)
|
||||
bus.stop()
|
||||
await asyncio.sleep(0.1)
|
||||
task.cancel()
|
||||
try:
|
||||
await task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
|
||||
assert len(received) == 1
|
||||
|
||||
_run(_test())
|
||||
|
||||
def test_no_subscriber_logs_warning(self):
|
||||
"""Messages to unsubscribed channels should warn, not crash."""
|
||||
|
||||
async def _test():
|
||||
bus = MessageBus()
|
||||
task = asyncio.create_task(bus.dispatch_outbound())
|
||||
await bus.publish_outbound(
|
||||
BusOutbound(
|
||||
channel="unknown",
|
||||
chat_id="c1",
|
||||
content="lost",
|
||||
)
|
||||
)
|
||||
await asyncio.sleep(0.1)
|
||||
bus.stop()
|
||||
await asyncio.sleep(0.1)
|
||||
task.cancel()
|
||||
try:
|
||||
await task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
|
||||
_run(_test())
|
||||
|
||||
def test_queue_sizes(self):
|
||||
async def _test():
|
||||
bus = MessageBus()
|
||||
@@ -1603,41 +1559,6 @@ class TestMessageBus:
|
||||
|
||||
_run(_test())
|
||||
|
||||
def test_subscriber_error_does_not_crash_dispatch(self):
|
||||
async def _test():
|
||||
bus = MessageBus()
|
||||
|
||||
async def bad_callback(msg):
|
||||
raise RuntimeError("subscriber crash")
|
||||
|
||||
bus.subscribe_outbound("tg", bad_callback)
|
||||
|
||||
task = asyncio.create_task(bus.dispatch_outbound())
|
||||
await bus.publish_outbound(
|
||||
BusOutbound(
|
||||
channel="tg",
|
||||
chat_id="c1",
|
||||
content="trigger",
|
||||
)
|
||||
)
|
||||
await asyncio.sleep(0.1)
|
||||
bus.stop()
|
||||
await asyncio.sleep(0.1)
|
||||
task.cancel()
|
||||
try:
|
||||
await task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
# dispatch should survive the error
|
||||
|
||||
_run(_test())
|
||||
|
||||
def test_stop_flag(self):
|
||||
bus = MessageBus()
|
||||
assert bus._running is False
|
||||
bus.stop()
|
||||
assert bus._running is False
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════
|
||||
# 9. Event dataclasses
|
||||
|
||||
Reference in New Issue
Block a user