Class FMQ

fastdeploy/inter_communicator/fmq.py:321–347

class FMQ:
    _instance = None
    _context = None

    def __new__(cls, config_path="fmq_config.json"):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            EndpointManager.load_config()

            # Determine IO threads based on global defaults
            io_threads = 1
            if EndpointManager.config.endpoints:
                # Use max io_threads among all endpoints
                io_threads = max(ep.io_threads for ep in EndpointManager.config.endpoints.values())

            cls._context = zmq.asyncio.Context(io_threads=io_threads)
        return cls._instance

    def queue(self, name: str, role="producer") -> Queue:
        return Queue(self._context, name, role)

    def topic(self, name: str) -> Topic:
        return Topic(self._context, name)

    async def destroy(self):
        # Destroy ZeroMQ context
        self._context.term()