MQTTBroker

async def publish_batch(
    self,
    *messages: "SendableMessage",
    queue: str,
) -> Any:
    msg = f"{self.__class__.__name__} doesn't support publishing in batches."
    raise FeatureNotSupportedException(msg)

def add_middleware(self, middleware: "BrokerMiddleware[Any, Any]") -> None:
    """Append BrokerMiddleware to the end of middlewares list.

    Current middleware will be used as a most inner of the stack.
    """
    self.config.add_middleware(middleware)

def insert_middleware(self, middleware: "BrokerMiddleware[Any, Any]") -> None:
    """Insert BrokerMiddleware to the start of middlewares list.

    Current middleware will be used as a most outer of the stack.
    """
    self.config.insert_middleware(middleware)

@override
def subscriber(  # type: ignore[override]
    self,
    topic: str,
    *,
    qos: QoS = QoS.AT_MOST_ONCE,
    shared: str | None = None,
    # broker arguments
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    dependencies: Iterable["Dependant"] = (),
    parser: Optional["CustomCallable"] = None,
    decoder: Optional["CustomCallable"] = None,
    max_workers: int = 1,
    persistent: bool = True,
    # AsyncAPI information
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
) -> "MQTTDefaultSubscriber | MQTTConcurrentSubscriber":
    """Subscribe a handler to an MQTT topic.

    Args:
        topic: MQTT topic filter. Wildcards ``+`` (single level) and
            ``#`` (multi-level) are supported.
        qos: QoS level for the subscription (0, 1, or 2).
        shared: Optional shared subscription group name. When set,
            subscribes as ``$share/<group>/<topic>``.
        ack_policy: Acknowledgement policy for message processing.
        no_reply: Whether to disable FastStream RPC / reply-to responses.
        dependencies: Dependencies list to apply to the subscriber.
        parser: Custom parser to map raw messages to FastStream ones.
        decoder: Function to decode FastStream message bytes to Python objects.
        max_workers: Number of workers to process messages concurrently.
        persistent: Whether to retain the subscriber across broker restarts.
        title: AsyncAPI subscriber object title.
        description: AsyncAPI subscriber object description.
        include_in_schema: Whether to include operation in AsyncAPI schema.
    """
    subscriber = create_subscriber(
        topic=topic,
        qos=qos,
        shared=shared,
        ack_policy=ack_policy,
        no_reply=no_reply,
        config=cast("MQTTBrokerConfig", self.config),
        max_workers=max_workers,
        title_=title,
        description_=description,
        include_in_schema=include_in_schema,
    )

    super().subscriber(subscriber, persistent=persistent)

    return subscriber.add_call(
        parser_=parser or self._parser,
        decoder_=decoder or self._decoder,
        dependencies_=dependencies,
    )

@override
def publisher(  # type: ignore[override]
    self,
    topic: str,
    *,
    qos: QoS = QoS.AT_MOST_ONCE,
    retain: bool = False,
    headers: dict[str, str] | None = None,
    persistent: bool = True,
    # AsyncAPI information
    title: str | None = None,
    description: str | None = None,
    schema: Any | None = None,
    include_in_schema: bool = True,
) -> "MQTTPublisher":
    """Create a persistent publisher object for the given MQTT topic.

    Args:
        topic: MQTT topic to publish to. Must not contain wildcards.
        qos: QoS level for published messages (0, 1, or 2).
        retain: Whether the broker should retain the last message.
        headers: Default headers to include in every published message.
        persistent: Whether to retain the publisher across broker restarts.
        title: AsyncAPI publisher object title.
        description: AsyncAPI publisher object description.
        schema: AsyncAPI publishing message type.
        include_in_schema: Whether to include operation in AsyncAPI schema.
    """
    publisher = create_publisher(
        topic=topic,
        qos=qos,
        retain=retain,
        headers=headers,
        broker_config=cast("MQTTBrokerConfig", self.config),
        title_=title,
        description_=description,
        schema_=schema,
        include_in_schema=include_in_schema,
    )
    super().publisher(publisher, persistent=persistent)
    return publisher

@override
def include_router(
    self,
    router: "MQTTRegistrator",  # type: ignore[override]
    *,
    prefix: str = "",
    dependencies: Iterable["Dependant"] = (),
    middlewares: Sequence["BrokerMiddleware[Any, Any]"] = (),
    include_in_schema: bool | None = None,
) -> None:
    if not isinstance(router, MQTTRegistrator):
        msg = (
            f"Router must be an instance of MQTTRegistrator, "
            f"got {type(router).__name__} instead."
        )
        raise TypeError(msg)

    super().include_router(
        router,
        prefix=prefix,
        dependencies=dependencies,
        middlewares=middlewares,
        include_in_schema=include_in_schema,
    )

    for m in router.config.broker_middlewares:
        router.config._validate_middleware(m)

def include_routers(
    self,
    *routers: "Registrator[MsgType, Any]",
) -> None:
    """Includes routers in the object."""
    for r in routers:
        self.include_router(r)

async def connect(self) -> ConnectionType:
    """Connect to a remote server."""
    if self._connection is None:
        self._connection = await self._connect()
        self._setup_logger()

    return self._connection

@override
async def start(self) -> None:
    await self.connect()
    c = MQTTBaseSubscriber.build_log_context(None, "")
    self.config.logger.log("Connection established", logging.INFO, c)
    await super().start()

@override
async def stop(
    self,
    exc_type: type[BaseException] | None = None,
    exc_val: BaseException | None = None,
    exc_tb: Optional["TracebackType"] = None,
) -> None:
    await super().stop(exc_type, exc_val, exc_tb)

    if self._connection is not None:
        await self._connection.disconnect()
        self._connection = None

    self.config.disconnect()

@override
async def ping(self, timeout: float | None = None) -> bool:
    if self._connection is None:
        return False
    try:
        await self._connection.ping(timeout=timeout or 5.0)
    except Exception:
        return False
    else:
        return True

@override
async def publish(
    self,
    message: "SendableMessage" = None,
    topic: str = "",
    *,
    qos: zmqtt.QoS = zmqtt.QoS.AT_MOST_ONCE,
    retain: bool = False,
    headers: dict[str, str] | None = None,
    correlation_id: str | None = None,
    reply_to: str = "",
) -> None:
    """Publish a message to an MQTT topic.

    Args:
        message: Message body to send.
        topic: MQTT topic to publish to.
        qos: QoS level (0, 1, or 2).
        retain: Whether the broker should retain the message.
        headers: Message headers (MQTT 5.0 user properties).
        correlation_id: Correlation ID for message tracing.
        reply_to: Response topic (MQTT 5.0 response_topic property).
    """
    cmd = MQTTPublishCommand(
        message,
        topic=topic,
        qos=qos,
        retain=retain,
        headers=headers,
        correlation_id=correlation_id or gen_cor_id(),
        reply_to=reply_to,
        _publish_type=PublishType.PUBLISH,
    )

    await self._basic_publish(cmd, producer=self.config.producer)

@override
async def request(
    self,
    message: "SendableMessage" = None,
    topic: str = "",
    /,
    timeout: float = 0.5,
    correlation_id: str | None = None,
    headers: dict[str, str] | None = None,
    qos: zmqtt.QoS = zmqtt.QoS.AT_MOST_ONCE,
    reply_to: str = "",
) -> "MQTTMessage":
    cmd = MQTTPublishCommand(
        message,
        topic=topic,
        correlation_id=correlation_id or gen_cor_id(),
        headers=headers,
        qos=qos,
        reply_to=reply_to,
        timeout=timeout,
        _publish_type=PublishType.REQUEST,
    )
    msg: MQTTMessage = await self._basic_request(cmd, producer=self.config.producer)
    return msg