| 1 | """Async message queue for decoupled channel-agent communication.""" |
| 2 | |
| 3 | import asyncio |
| 4 | |
| 5 | from nanobot.bus.events import InboundMessage, OutboundMessage, RuntimeEvent |
| 6 | |
| 7 | |
| 8 | class MessageBus: |
| 9 | """ |
| 10 | Async message bus that decouples chat channels from the agent core. |
| 11 | |
| 12 | Channels push messages to the inbound queue, and the agent processes |
| 13 | them and pushes responses to the outbound queue. |
| 14 | """ |
| 15 | |
| 16 | def __init__(self): |
| 17 | self.inbound: asyncio.Queue[InboundMessage] = asyncio.Queue() |
| 18 | self.outbound: asyncio.Queue[OutboundMessage] = asyncio.Queue() |
| 19 | self.runtime: asyncio.Queue[RuntimeEvent] = asyncio.Queue(maxsize=1000) |
| 20 | |
| 21 | async def publish_inbound(self, msg: InboundMessage) -> None: |
| 22 | """Publish a message from a channel to the agent.""" |
| 23 | await self.inbound.put(msg) |
| 24 | |
| 25 | async def consume_inbound(self) -> InboundMessage: |
| 26 | """Consume the next inbound message (blocks until available).""" |
| 27 | return await self.inbound.get() |
| 28 | |
| 29 | async def publish_outbound(self, msg: OutboundMessage) -> None: |
| 30 | """Publish a response from the agent to channels.""" |
| 31 | await self.outbound.put(msg) |
| 32 | |
| 33 | async def consume_outbound(self) -> OutboundMessage: |
| 34 | """Consume the next outbound message (blocks until available).""" |
| 35 | return await self.outbound.get() |
| 36 | |
| 37 | async def publish_runtime(self, event: RuntimeEvent) -> None: |
| 38 | """Publish an internal runtime event that is not consumed by the agent loop.""" |
| 39 | if self.runtime.full(): |
| 40 | try: |
| 41 | self.runtime.get_nowait() |
| 42 | except asyncio.QueueEmpty: |
| 43 | pass |
| 44 | await self.runtime.put(event) |
| 45 | |
| 46 | async def consume_runtime(self) -> RuntimeEvent: |
| 47 | """Consume the next runtime event (blocks until available).""" |
| 48 | return await self.runtime.get() |
| 49 | |
| 50 | @property |
| 51 | def inbound_size(self) -> int: |
| 52 | """Number of pending inbound messages.""" |
| 53 | return self.inbound.qsize() |
| 54 | |
| 55 | @property |
| 56 | def outbound_size(self) -> int: |
| 57 | """Number of pending outbound messages.""" |
| 58 | return self.outbound.qsize() |
| 59 | |
| 60 | @property |
| 61 | def runtime_size(self) -> int: |
| 62 | """Number of pending internal runtime events.""" |
| 63 | return self.runtime.qsize() |
| 64 |