orchestrata_ultimate.postman.queue_manager

Reference › orchestrata_ultimate › postman › queue_manager

The registry of named queues for the broker, SPEC-002 story US-003 (REQ-002-US-003-001 to -009). QueueManager implements IQueueManager from interfaces and builds queues from a QueueConfig using the backends in queue. delivery imports it, and uses it to give each recipient a delivery queue.

create_queue(name, config) raises QueueError if the name exists. Otherwise it builds a DiskBackedQueue (stored as <name>.queue under the storage directory, default /var/lib/postman) when persistence is enabled and an InMemoryQueue when not, wraps the result in a BoundedQueue when the config sets a positive maximum size, and records it. get_queue, list_queues, delete_queue, purge_queue and get_status (name and current size) are straightforward. get_dead_letter_queue creates, on first use and under a lock, the dead-letter queue named _dead_letter: persistent and unbounded. close_all closes every queue. A single asyncio lock guards the tables.

At a glance

Class diagram

Dependency diagram

API

Classes

QueueManager

Bases: IQueueManager · Decorators: traces(spec='002', story='US-003', reqs=['REQ-002-US-003-001', 'REQ-002-US-003-002', 'REQ-002-US-003-003', 'REQ-002-US-… · Line 42

Manages multiple queues with different configurations.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/postman/queue_manager.py

# mypy: ignore-errors
"""
Queue Manager for Postman Message Broker.

Manages creation, deletion, and lifecycle of message queues.
Supports in-memory and disk-backed queues with dead-letter queue support.
"""

import asyncio
import logging
from pathlib import Path

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.postman.constants import (
    DEFAULT_DEAD_LETTER_QUEUE_NAME,
    DEFAULT_STORAGE_PATH,
)
from orchestrata_ultimate.postman.exceptions import QueueError, QueueNotFoundError
from orchestrata_ultimate.postman.interfaces import IQueue, IQueueManager
from orchestrata_ultimate.postman.queue import (
    BoundedQueue,
    DiskBackedQueue,
    InMemoryQueue,
)
from orchestrata_ultimate.postman.types import QueueConfig, QueueStatus

logger = logging.getLogger(__name__)


@traces(
    spec="002",
    story="US-003",
    reqs=[
        "REQ-002-US-003-001",
        "REQ-002-US-003-002",
        "REQ-002-US-003-003",
        "REQ-002-US-003-004",
        "REQ-002-US-003-005",
        "REQ-002-US-003-009",
    ],
)
class QueueManager(IQueueManager):
    """
    Manages multiple queues with different configurations.

    Features:
    - Create/delete named queues
    - Support for in-memory and disk-backed queues
    - Bounded queue support via max_size configuration
    - Dead-letter queue for failed messages
    - Thread-safe queue management
    - Queue status and metrics tracking
    """

    @traces(spec="002", story="US-009", reqs=["REQ-002-US-009-012"])
    def __init__(self, storage_path: Path | None = None) -> None:
        """
        Initialize queue manager.

        Args:
            storage_path: Base directory for persistent queues
        """
        self._queues: dict[str, IQueue] = {}
        self._configs: dict[str, QueueConfig] = {}
        self._lock = asyncio.Lock()
        self._storage_path = storage_path or Path(DEFAULT_STORAGE_PATH)
        self._dead_letter_queue: IQueue | None = None

    @traces(
        spec="002",
        story="US-003",
        reqs=["REQ-002-US-003-001", "REQ-002-FR4-004"],
    )
    async def create_queue(self, name: str, config: QueueConfig) -> IQueue:
        """
        Create a new queue with specified configuration.

        Args:
            name: Queue name
            config: Queue configuration

        Returns:
            Created queue instance

        Raises:
            QueueError: If queue already exists
        """
        async with self._lock:
            if name in self._queues:
                raise QueueError(f"Queue '{name}' already exists")

            # Create base queue based on persistence setting
            if config.persistence_enabled:
                queue_path = self._storage_path / f"{name}.queue"
                # DiskBackedQueue.__init__ takes storage_path: str.
                base_queue = DiskBackedQueue(name=name, storage_path=str(queue_path))
            else:
                base_queue = InMemoryQueue(name=name)

            # Wrap with bounded queue if max_size specified
            if config.max_size and config.max_size > 0:
                queue = BoundedQueue(base_queue=base_queue, max_size=config.max_size)
            else:
                queue = base_queue

            # Store queue and config
            self._queues[name] = queue
            self._configs[name] = config

            return queue

    @traces(spec="002", story="US-003", reqs=["REQ-002-US-003-002"])
    async def delete_queue(self, name: str) -> bool:
        """
        Delete a queue by name.

        Args:
            name: Queue name

        Returns:
            True if deleted, False if not found

        Raises:
            QueueError: If attempting to delete dead-letter queue
        """
        async with self._lock:
            if name == DEFAULT_DEAD_LETTER_QUEUE_NAME:
                raise QueueError("Cannot delete dead-letter queue")

            if name not in self._queues:
                return False

            queue = self._queues[name]

            # Clean up disk-backed queue storage if applicable
            if isinstance(queue, BoundedQueue):
                # Unwrap to get base queue
                if hasattr(queue, "_queue") and isinstance(
                    queue._queue, DiskBackedQueue
                ):
                    await queue._queue.close()
            elif isinstance(queue, DiskBackedQueue):
                await queue.close()

            # Remove from tracking
            del self._queues[name]
            del self._configs[name]

            return True

    async def get_queue(self, name: str) -> IQueue | None:
        """
        Get queue by name.

        Args:
            name: Queue name

        Returns:
            Queue instance or None if not found
        """
        async with self._lock:
            return self._queues.get(name)

    @traces(spec="002", story="US-003", reqs=["REQ-002-US-003-003"])
    async def list_queues(self) -> list[str]:
        """
        List all queue names.

        Returns:
            List of queue names
        """
        async with self._lock:
            return list(self._queues.keys())

    @traces(spec="002", story="US-003", reqs=["REQ-002-US-003-004"])
    async def get_status(self, name: str) -> QueueStatus | None:
        """
        Get queue status and metrics.

        Args:
            name: Queue name

        Returns:
            Queue status or None if not found
        """
        async with self._lock:
            queue = self._queues.get(name)
            if not queue:
                return None

            # Build status matching QueueStatus dataclass
            size = await queue.size()

            return QueueStatus(
                name=name,
                size=size,
            )

    @traces(spec="002", story="US-003", reqs=["REQ-002-US-003-005"])
    async def purge_queue(self, name: str) -> int:
        """
        Clear all messages from queue.

        Args:
            name: Queue name

        Returns:
            Number of messages purged

        Raises:
            QueueNotFoundError: If queue not found
        """
        async with self._lock:
            queue = self._queues.get(name)
            if not queue:
                raise QueueNotFoundError(f"Queue '{name}' not found")

            purged_count = 0
            while not await queue.is_empty():
                await queue.dequeue()
                purged_count += 1

            return purged_count

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-002-027"])
    @traces(spec="002", story="US-003", reqs=["REQ-002-US-003-009"])
    async def get_dead_letter_queue(self) -> IQueue:
        """
        Get or create dead-letter queue.

        The DLQ is automatically created on first access with:
        - Persistence enabled
        - No size limit
        - Name from DEFAULT_DEAD_LETTER_QUEUE_NAME

        Returns:
            Dead-letter queue instance
        """
        if self._dead_letter_queue is None:
            async with self._lock:
                # Double-check after acquiring lock
                if self._dead_letter_queue is None:
                    dlq_name = DEFAULT_DEAD_LETTER_QUEUE_NAME

                    # Check if already exists in managed queues
                    if dlq_name in self._queues:
                        self._dead_letter_queue = self._queues[dlq_name]
                    else:
                        # Create persistent DLQ with no size limit
                        dlq_config = QueueConfig(
                            name=dlq_name,
                            persistence_enabled=True,
                            max_size=0,  # 0 means no limit
                        )

                        # Create queue directly to avoid recursive lock
                        queue_path = self._storage_path / f"{dlq_name}.queue"
                        # DiskBackedQueue.__init__ takes storage_path: str.
                        self._dead_letter_queue = DiskBackedQueue(
                            name=dlq_name, storage_path=str(queue_path)
                        )
                        self._queues[dlq_name] = self._dead_letter_queue
                        self._configs[dlq_name] = dlq_config

        return self._dead_letter_queue

    async def close_all(self) -> None:
        """
        Close all queues and cleanup resources.

        This method is not part of IQueueManager but useful for cleanup.
        """
        async with self._lock:
            for name, queue in list(self._queues.items()):
                # Close disk-backed queues
                if isinstance(queue, BoundedQueue):
                    if hasattr(queue, "_queue") and isinstance(
                        queue._queue, DiskBackedQueue
                    ):
                        await queue._queue.close()
                elif isinstance(queue, DiskBackedQueue):
                    await queue.close()

            self._queues.clear()
            self._configs.clear()
            self._dead_letter_queue = None