RabbitBroker

def setup(self) -> None:
    """Prepare all Broker entities to startup."""
    for h in self._subscribers.values():
        self.setup_subscriber(h)

    for p in self._publishers.values():
        self.setup_publisher(p)

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

    Current middleware will be used as a most inner of already existed ones.
    """
    self._middlewares = (*self._middlewares, middleware)

    for sub in self._subscribers.values():
        sub.add_middleware(middleware)

    for pub in self._publishers.values():
        pub.add_middleware(middleware)

@override
def subscriber(  # type: ignore[override]
    self,
    queue: Union[str, "RabbitQueue"],
    exchange: Union[str, "RabbitExchange", None] = None,
    *,
    consume_args: Optional["AnyDict"] = None,
    dependencies: Iterable["Depends"] = (),
    parser: Optional["CustomCallable"] = None,
    decoder: Optional["CustomCallable"] = None,
    middlewares: Sequence["SubscriberMiddleware[RabbitMessage]"] = (),
    channel: Optional["Channel"] = None,
    reply_config: Annotated[
        Optional["ReplyConfig"],
        deprecated(
            "Deprecated in **FastStream 0.5.16**. "
            "Please, use `RabbitResponse` object as a handler return instead. "
            "Argument will be removed in **FastStream 0.6.0**."
        ),
    ] = None,
    filter: Annotated[
        "Filter[RabbitMessage]",
        deprecated(
            "Deprecated in **FastStream 0.5.0**. Please, create `subscriber` object "
            "and use it explicitly instead. Argument will be removed in **FastStream 0.6.0**."
        ),
    ] = default_filter,
    retry: Union[bool, int] = False,
    no_ack: bool = False,
    no_reply: bool = False,
    title: Optional[str] = None,
    description: Optional[str] = None,
    include_in_schema: bool = True,
) -> AsyncAPISubscriber:
    """Declares RabbitMQ subscriber object and binds it to the exchange.

    You can use it as a handler decorator - `@broker.subscriber(...)`.
    Or you can create a subscriber object to call it lately - `broker.subscriber(...)`.

    Args:
        queue: RabbitMQ queue to listen. **FastStream** declares and binds
        queue object to `exchange` automatically if it is not passive (by default).
        exchange: RabbitMQ exchange to bind queue to. Uses default exchange
        if not presented. **FastStream** declares exchange object automatically
        if it is not passive (by default).
        consume_args: Extra consumer arguments to use in `queue.consume(...)` method.
        channel: Channel to use for consuming messages. If not specified, a default channel will be used.
        reply_config: Extra options to use at replies publishing.
        dependencies: Dependencies list (`[Depends(),]`) to apply to the subscriber.
        parser: Parser to map original **IncomingMessage** Msg to FastStream one.
        decoder: Function to decode FastStream msg bytes body to python objects.
        middlewares: Subscriber middlewares to wrap incoming message processing.
        filter: Overload subscriber to consume various messages from the same source.
        retry: Whether to `nack` message at processing exception.
        no_ack: Whether to disable **FastStream** autoacknowledgement logic or not.
        no_reply: Whether to disable **FastStream** RPC and Reply To auto responses or not.
        title: AsyncAPI subscriber object title.
        description: AsyncAPI subscriber object description. Uses decorated docstring as default.
        include_in_schema: Whether to include operation in AsyncAPI schema or not.
    """
    subscriber = cast(
        "AsyncAPISubscriber",
        super().subscriber(
            create_subscriber(
                queue=RabbitQueue.validate(queue),
                exchange=RabbitExchange.validate(exchange),
                consume_args=consume_args,
                reply_config=reply_config,
                channel=channel,
                # subscriber args
                no_ack=no_ack,
                no_reply=no_reply,
                retry=retry,
                broker_middlewares=self._middlewares,
                broker_dependencies=self._dependencies,
                # AsyncAPI
                title_=title,
                description_=description,
                include_in_schema=self._solve_include_in_schema(include_in_schema),
            )
        ),
    )

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

@override
def publisher(  # type: ignore[override]
    self,
    queue: Union["RabbitQueue", str] = "",
    exchange: Union["RabbitExchange", str, None] = None,
    *,
    routing_key: str = "",
    mandatory: bool = True,
    immediate: bool = False,
    timeout: "TimeoutType" = None,
    persist: bool = False,
    reply_to: Optional[str] = None,
    priority: Optional[int] = None,
    middlewares: Sequence["PublisherMiddleware"] = (),
    title: Optional[str] = None,
    description: Optional[str] = None,
    schema: Optional[Any] = None,
    include_in_schema: bool = True,
    headers: Optional["HeadersType"] = None,
    content_type: Optional[str] = None,
    content_encoding: Optional[str] = None,
    expiration: Optional["DateType"] = None,
    message_type: Optional[str] = None,
    user_id: Optional[str] = None,
) -> AsyncAPIPublisher:
    """Creates long-living and AsyncAPI-documented publisher object.

    You can use it as a handler decorator (handler should be decorated by `@broker.subscriber(...)` too) - `@broker.publisher(...)`.
    In such case publisher will publish your handler return value.

    Or you can create a publisher object to call it lately - `broker.publisher(...).publish(...)`.

    Args:
        queue: Default message routing key to publish with.
        exchange: Target exchange to publish message to.
        routing_key: Default message routing key to publish with.
        Overrides `queue` option if presented.
        mandatory: Client waits for confirmation that the message is placed
            to some queue. RabbitMQ returns message to client if there is no suitable queue.
        immediate: Client expects that there is a consumer ready to take the message to work.
            RabbitMQ returns message to client if there is no suitable consumer.
        timeout: Send confirmation time from RabbitMQ.
        persist: Restore the message on RabbitMQ reboot.
        reply_to: Reply message routing key to send with (always sending to default exchange).
        priority: The message priority (0 by default).
        middlewares: Publisher middlewares to wrap outgoing messages.
        title: AsyncAPI publisher object title.
        description: AsyncAPI publisher object description.
        schema: AsyncAPI publishing message type. Should be any python-native
            object annotation or `pydantic.BaseModel`.
        include_in_schema: Whether to include operation in AsyncAPI schema or not.
        headers: Message headers to store meta-information. Can be overridden
            by `publish.headers` if specified.
        content_type: Message **content-type** header. Used by application, not core RabbitMQ.
            Will be set automatically if not specified.
        content_encoding: Message body content encoding, e.g. **gzip**.
        expiration: Message expiration (lifetime) in seconds (or datetime or timedelta).
        message_type: Application-specific message type, e.g. **orders.created**.
        user_id: Publisher connection User ID, validated if set.
    """
    message_kwargs = PublishKwargs(
        mandatory=mandatory,
        immediate=immediate,
        timeout=timeout,
        persist=persist,
        reply_to=reply_to,
        headers=headers,
        priority=priority,
        content_type=content_type,
        content_encoding=content_encoding,
        message_type=message_type,
        user_id=user_id,
        expiration=expiration,
    )

    publisher = cast(
        "AsyncAPIPublisher",
        super().publisher(
            AsyncAPIPublisher.create(
                routing_key=routing_key,
                queue=RabbitQueue.validate(queue),
                exchange=RabbitExchange.validate(exchange),
                message_kwargs=message_kwargs,
                # Specific
                broker_middlewares=self._middlewares,
                middlewares=middlewares,
                # AsyncAPI
                title_=title,
                description_=description,
                schema_=schema,
                include_in_schema=self._solve_include_in_schema(include_in_schema),
            )
        ),
    )

    return publisher

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

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

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

def get_fmt(self) -> str:
    return (
        "%(asctime)s %(levelname)-8s - "
        f"%(exchange)-{self._max_exchange_len}s | "
        f"%(queue)-{self._max_queue_len}s | "
        f"%(message_id)-{self.__max_msg_id_ln}s "
        "- %(message)s"
    )

def setup_subscriber(
    self,
    subscriber: SubscriberProto[MsgType],
    **kwargs: Any,
) -> None:
    """Setup the Subscriber to prepare it to starting."""
    data = self._subscriber_setup_extra.copy()
    data.update(kwargs)
    subscriber.setup(**data)

def setup_publisher(
    self,
    publisher: "PublisherProto[MsgType]",
    **kwargs: Any,
) -> None:
    """Setup the Publisher to prepare it to starting."""
    data = self._publisher_setup_extra.copy()
    data.update(kwargs)
    publisher.setup(**data)

async def close(
    self,
    exc_type: Optional[Type[BaseException]] = None,
    exc_val: Optional[BaseException] = None,
    exc_tb: Optional["TracebackType"] = None,
) -> None:
    """Closes the object."""
    self.running = False

    for h in self._subscribers.values():
        await h.close()

    if self._connection is not None:
        await self._close(exc_type, exc_val, exc_tb)

@override
async def connect(  # type: ignore[override]
    self,
    url: Union[str, "URL", None] = EMPTY,
    *,
    host: Optional[str] = None,
    port: Optional[int] = None,
    virtualhost: Optional[str] = None,
    ssl_options: Optional["SSLOptions"] = None,
    client_properties: Optional["RabbitClientProperties"] = None,
    security: Optional["BaseSecurity"] = None,
    timeout: "TimeoutType" = None,
    fail_fast: bool = EMPTY,
    reconnect_interval: "TimeoutType" = EMPTY,
    # channel args
    default_channel: Optional[Channel] = None,
    channel_number: Annotated[
        Optional[int],
        deprecated(
            "Deprecated in **FastStream 0.5.39**. "
            "Please, use `default_channel=Channel(channel_number=...)` instead. "
            "Argument will be removed in **FastStream 0.6.0**."
        ),
    ] = EMPTY,
    publisher_confirms: Annotated[
        bool,
        deprecated(
            "Deprecated in **FastStream 0.5.39**. "
            "Please, use `default_channel=Channel(publisher_confirms=...)` instead. "
            "Argument will be removed in **FastStream 0.6.0**."
        ),
    ] = EMPTY,
    on_return_raises: Annotated[
        bool,
        deprecated(
            "Deprecated in **FastStream 0.5.39**. "
            "Please, use `default_channel=Channel(on_return_raises=...)` instead. "
            "Argument will be removed in **FastStream 0.6.0**."
        ),
    ] = EMPTY,
) -> "RobustConnection":
    """Connect broker object to RabbitMQ.

    To startup subscribers too you should use `broker.start()` after/instead this method.

    Args:
        url: RabbitMQ destination location to connect.
        host: Destination host. This option overrides `url` option host.
        port: Destination port. This option overrides `url` option port.
        virtualhost: RabbitMQ virtual host to use in the current broker connection.
        ssl_options: Extra ssl options to establish connection.
        client_properties: Add custom client capability.
        security: Security options to connect broker and generate AsyncAPI server security information.
        timeout: Connection establishement timeout.
        fail_fast: Broker startup raises `AMQPConnectionError` if RabbitMQ is unreachable.
        reconnect_interval: Time to sleep between reconnection attempts.
        default_channel: Default channel settings to use.
        channel_number: Specify the channel number explicit.
        publisher_confirms: if `True` the `publish` method will
            return `bool` type after publish is complete.
            Otherwise it will returns `None`.
        on_return_raises: raise an :class:`aio_pika.exceptions.DeliveryError`
            when mandatory message will be returned
    """
    kwargs: AnyDict = {}

    if not default_channel and (
        channel_number is not EMPTY
        or publisher_confirms is not EMPTY
        or on_return_raises is not EMPTY
    ):
        default_channel = Channel(
            channel_number=None if channel_number is EMPTY else channel_number,
            publisher_confirms=True
            if publisher_confirms is EMPTY
            else publisher_confirms,
            on_return_raises=False
            if on_return_raises is EMPTY
            else on_return_raises,
        )

    if default_channel:
        kwargs["channel_settings"] = default_channel

    if timeout:
        kwargs["timeout"] = timeout

    if fail_fast is not EMPTY:
        kwargs["fail_fast"] = fail_fast

    if reconnect_interval is not EMPTY:
        kwargs["reconnect_interval"] = reconnect_interval

    url = None if url is EMPTY else url

    if url or any(
        (host, port, virtualhost, ssl_options, client_properties, security)
    ):
        security_args = parse_security(security)

        kwargs["url"] = build_url(
            url,
            host=host,
            port=port,
            virtualhost=virtualhost,
            ssl_options=ssl_options,
            client_properties=client_properties,
            login=security_args.get("login"),
            password=security_args.get("password"),
            ssl=security_args.get("ssl"),
        )

        if ssl_context := security_args.get("ssl_context"):
            kwargs["ssl_context"] = ssl_context

    connection = await super().connect(**kwargs)

    return connection

async def start(self) -> None:
    """Connect broker to RabbitMQ and startup all subscribers."""
    await super().start()

    assert self.declarer, NOT_CONNECTED_YET  # nosec B101

    for publisher in self._publishers.values():
        if publisher.exchange is not None:
            await self.declare_exchange(publisher.exchange)

    for subscriber in self._subscribers.values():
        self._log(
            f"`{subscriber.call_name}` waiting for messages",
            extra=subscriber.get_log_context(None),
        )
        await subscriber.start()

@override
async def publish(  # type: ignore[override]
    self,
    message: "AioPikaSendableMessage" = None,
    queue: Union["RabbitQueue", str] = "",
    exchange: Union["RabbitExchange", str, None] = None,
    *,
    routing_key: str = "",
    mandatory: bool = True,
    immediate: bool = False,
    timeout: "TimeoutType" = None,
    persist: bool = False,
    reply_to: Optional[str] = None,
    rpc: Annotated[
        bool,
        deprecated(
            "Deprecated in **FastStream 0.5.17**. Please, use `request` method instead. "
            "Argument will be removed in **FastStream 0.6.0**."
        ),
    ] = False,
    rpc_timeout: Annotated[
        Optional[float],
        deprecated(
            "Deprecated in **FastStream 0.5.17**. Please, use `request` method with `timeout` instead. "
            "Argument will be removed in **FastStream 0.6.0**."
        ),
    ] = 30.0,
    raise_timeout: Annotated[
        bool,
        deprecated(
            "Deprecated in **FastStream 0.5.17**. `request` always raises TimeoutError instead. "
            "Argument will be removed in **FastStream 0.6.0**."
        ),
    ] = False,
    # message args
    correlation_id: Optional[str] = None,
    headers: Optional["HeadersType"] = None,
    content_type: Optional[str] = None,
    content_encoding: Optional[str] = None,
    expiration: Optional["DateType"] = None,
    message_id: Optional[str] = None,
    timestamp: Optional["DateType"] = None,
    message_type: Optional[str] = None,
    user_id: Optional[str] = None,
    priority: Optional[int] = None,
) -> Optional[Any]:
    """Publish message directly.

    This method allows you to publish message in not AsyncAPI-documented way. You can use it in another frameworks
    applications or to publish messages from time to time.

    Please, use `@broker.publisher(...)` or `broker.publisher(...).publish(...)` instead in a regular way.

    Args:
        message: Message body to send.
        queue: Message routing key to publish with.
        exchange: Target exchange to publish message to.
        routing_key: Message routing key to publish with. Overrides `queue` option if presented.
        mandatory: Client waits for confirmation that the message is placed to some queue.
        RabbitMQ returns message to client if there is no suitable queue.
        immediate: Client expects that there is consumer ready to take the message to work.
        RabbitMQ returns message to client if there is no suitable consumer.
        timeout: Send confirmation time from RabbitMQ.
        persist: Restore the message on RabbitMQ reboot.
        reply_to: Reply message routing key to send with (always sending to default exchange).
        rpc: Whether to wait for reply in blocking mode.
        rpc_timeout: RPC reply waiting time.
        raise_timeout: Whether to raise `TimeoutError` or return `None` at **rpc_timeout**.
        correlation_id: Manual message **correlation_id** setter.
        **correlation_id** is a useful option to trace messages.
        headers: Message headers to store metainformation.
        content_type: Message **content-type** header.
        Used by application, not core RabbitMQ. Will be set automatically if not specified.
        content_encoding: Message body content encoding, e.g. **gzip**.
        expiration: Message expiration (lifetime) in seconds (or datetime or timedelta).
        message_id: Arbitrary message id. Generated automatically if not presented.
        timestamp: Message publish timestamp. Generated automatically if not presented.
        message_type: Application-specific message type, e.g. **orders.created**.
        user_id: Publisher connection User ID, validated if set.
        priority: The message priority (0 by default).
    """
    routing = routing_key or RabbitQueue.validate(queue).routing
    correlation_id = correlation_id or gen_cor_id()

    return await super().publish(
        message,
        producer=self._producer,
        routing_key=routing,
        app_id=self.app_id,
        exchange=exchange,
        mandatory=mandatory,
        immediate=immediate,
        persist=persist,
        reply_to=reply_to,
        headers=headers,
        correlation_id=correlation_id,
        content_type=content_type,
        content_encoding=content_encoding,
        expiration=expiration,
        message_id=message_id,
        timestamp=timestamp,
        message_type=message_type,
        user_id=user_id,
        timeout=timeout,
        priority=priority,
        rpc=rpc,
        rpc_timeout=rpc_timeout,
        raise_timeout=raise_timeout,
    )