Class MemoryQueue

methods/MIRIX/mirix/queue/memory_queue.py:18–49

Async in-memory queue implementation (single partition)

class MemoryQueue(QueueInterface):
    """Async in-memory queue implementation (single partition)"""

    def __init__(self):
        self._queue: asyncio.Queue[QueueMessage] = asyncio.Queue()

    async def put(self, message: QueueMessage) -> None:
        logger.debug("Adding message to queue: agent_id=%s", message.agent_id)
        await self._queue.put(message)

    async def get(self, timeout: Optional[float] = None) -> QueueMessage:
        """
        Retrieve a message from the queue.

        Args:
            timeout: Optional timeout in seconds (None = block indefinitely)

        Returns:
            QueueMessage from the queue

        Raises:
            asyncio.TimeoutError: If no message available within timeout
        """
        if timeout is not None:
            message = await asyncio.wait_for(self._queue.get(), timeout=timeout)
        else:
            message = await self._queue.get()
        logger.debug("Retrieved message from queue: agent_id=%s", message.agent_id)
        return message

    async def close(self) -> None:
        pass


class PartitionedMemoryQueue(QueueInterface):