Method _create_queue

methods/MIRIX/mirix/queue/manager.py:127–171

Factory method to create the appropriate queue implementation.

(self)

        logger.info("Queue manager initialized successfully")

    def _create_queue(self) -> QueueInterface:
        """Factory method to create the appropriate queue implementation."""
        if config.QUEUE_TYPE == "kafka":
            try:
                from .kafka_queue import KafkaQueue

                kafka_kwargs = {
                    "bootstrap_servers": config.KAFKA_BOOTSTRAP_SERVERS,
                    "topic": config.KAFKA_TOPIC,
                    "group_id": config.KAFKA_GROUP_ID,
                    "serialization_format": config.KAFKA_SERIALIZATION_FORMAT,
                    "security_protocol": config.KAFKA_SECURITY_PROTOCOL,
                    "auto_offset_reset": config.KAFKA_AUTO_OFFSET_RESET,
                    "consumer_timeout_ms": config.KAFKA_CONSUMER_TIMEOUT_MS,
                    "max_poll_interval_ms": config.KAFKA_MAX_POLL_INTERVAL_MS,
                    "session_timeout_ms": config.KAFKA_SESSION_TIMEOUT_MS,
                }

                if config.KAFKA_SSL_CAFILE:
                    kafka_kwargs["ssl_cafile"] = config.KAFKA_SSL_CAFILE
                if config.KAFKA_SSL_CERTFILE:
                    kafka_kwargs["ssl_certfile"] = config.KAFKA_SSL_CERTFILE
                if config.KAFKA_SSL_KEYFILE:
                    kafka_kwargs["ssl_keyfile"] = config.KAFKA_SSL_KEYFILE

                return KafkaQueue(**kafka_kwargs)
            except ImportError as e:
                raise ImportError(
                    f"Kafka queue requested but dependencies not installed: {e}\n"
                    "Install with: pip install queue-sample[kafka]"
                ) from e
        else:
            if self._num_workers > 1:
                mode = "round-robin" if self._round_robin else "hash"
                logger.info(
                    "Using PartitionedMemoryQueue with %d partitions, mode=%s",
                    self._num_workers,
                    mode,
                )
                return PartitionedMemoryQueue(
                    num_partitions=self._num_workers,
                    round_robin=self._round_robin,
                )
            else:
                return MemoryQueue()

    async def save(self, message: QueueMessage) -> None:
        """