AsyncAPISubscriber

@override
def setup(  # type: ignore[override]
    self,
    *,
    app_id: Optional[str],
    virtual_host: str,
    declarer: "RabbitDeclarer",
    # basic args
    logger: Optional["LoggerProto"],
    producer: Optional["AioPikaFastProducer"],
    graceful_timeout: Optional[float],
    extra_context: "AnyDict",
    # broker options
    broker_parser: Optional["CustomCallable"],
    broker_decoder: Optional["CustomCallable"],
    # dependant args
    apply_types: bool,
    is_validate: bool,
    _get_dependant: Optional[Callable[..., Any]],
    _call_decorators: Iterable["Decorator"],
) -> None:
    self.app_id = app_id
    self.virtual_host = virtual_host
    self.declarer = declarer

    super().setup(
        logger=logger,
        producer=producer,
        graceful_timeout=graceful_timeout,
        extra_context=extra_context,
        broker_parser=broker_parser,
        broker_decoder=broker_decoder,
        apply_types=apply_types,
        is_validate=is_validate,
        _get_dependant=_get_dependant,
        _call_decorators=_call_decorators,
    )

def add_prefix(self, prefix: str) -> None:
    """Include Subscriber in router."""
    self.queue = self.queue.add_prefix(prefix)

def schema(self) -> Dict[str, Channel]:
    """Returns the schema of the API operation as a dictionary of channel names and channel objects."""
    if self.include_in_schema:
        return self.get_schema()
    else:
        return {}

def add_middleware(self, middleware: "BrokerMiddleware[MsgType]") -> None:
    self._broker_middlewares = (*self._broker_middlewares, middleware)

def get_log_context(
    self,
    message: Optional["StreamMessage[Any]"],
) -> Dict[str, str]:
    return self.build_log_context(
        message=message,
        queue=self.queue,
        exchange=self.exchange,
    )

@override
async def start(self) -> None:
    """Starts the consumer for the RabbitMQ queue."""
    if self.declarer is None:
        raise SetupError("You should setup subscriber at first.")

    self._queue_obj = queue = await self.declarer.declare_queue(self.queue)

    if (
        self.exchange is not None
        and not queue.passive  # queue just getted from RMQ
        and self.exchange.name  # check Exchange is not default
    ):
        exchange = await self.declarer.declare_exchange(self.exchange)

        await queue.bind(
            exchange,
            routing_key=self.queue.routing,
            arguments=self.queue.bind_arguments,
            timeout=self.queue.timeout,
            robust=self.queue.robust,
        )

    if self.calls:
        self._consumer_tag = await self._queue_obj.consume(
            # NOTE: aio-pika expects AbstractIncomingMessage, not IncomingMessage
            self.consume,  # type: ignore[arg-type]
            arguments=self.consume_args,
        )

    await super().start()

async def close(self) -> None:
    await super().close()

    if self._queue_obj is not None:
        if self._consumer_tag is not None:  # pragma: no branch
            if not self._queue_obj.channel.is_closed:
                await self._queue_obj.cancel(self._consumer_tag)
            self._consumer_tag = None

        self._queue_obj = None

async def consume(self, msg: MsgType) -> Any:
    """Consume a message asynchronously."""
    if not self.running:
        return None

    try:
        return await self.process_message(msg)

    except StopConsume:
        # Stop handler at StopConsume exception
        await self.close()

    except SystemExit:
        # Stop handler at `exit()` call
        await self.close()

        if app := context.get("app"):
            app.exit()

    except Exception:  # nosec B110
        # All other exceptions were logged by CriticalLogMiddleware
        pass

async def process_message(self, msg: MsgType) -> "Response":
    """Execute all message processing stages."""
    async with AsyncExitStack() as stack:
        stack.enter_context(self.lock)

        # Enter context before middlewares
        for k, v in self.extra_context.items():
            stack.enter_context(context.scope(k, v))

        stack.enter_context(context.scope("handler_", self))

        # enter all middlewares
        middlewares: List[BaseMiddleware] = []
        for base_m in self._broker_middlewares:
            middleware = base_m(msg)
            middlewares.append(middleware)
            await middleware.__aenter__()

        cache: Dict[Any, Any] = {}
        parsing_error: Optional[Exception] = None
        for h in self.calls:
            try:
                message = await h.is_suitable(msg, cache)
            except Exception as e:
                parsing_error = e
                break

            if message is not None:
                # Acknowledgement scope
                # TODO: move it to scope enter at `retry` option deprecation
                await stack.enter_async_context(
                    self.watcher(
                        message,
                        **self.extra_watcher_options,
                    )
                )

                stack.enter_context(
                    context.scope("log_context", self.get_log_context(message))
                )
                stack.enter_context(context.scope("message", message))

                # Middlewares should be exited before scope release
                for m in middlewares:
                    stack.push_async_exit(m.__aexit__)

                result_msg = ensure_response(
                    await h.call(
                        message=message,
                        # consumer middlewares
                        _extra_middlewares=(m.consume_scope for m in middlewares),
                    )
                )

                if not result_msg.correlation_id:
                    result_msg.correlation_id = message.correlation_id

                for p in chain(
                    self.__get_response_publisher(message),
                    h.handler._publishers,
                ):
                    await p.publish(
                        result_msg.body,
                        **result_msg.as_publish_kwargs(),
                        # publisher middlewares
                        _extra_middlewares=(m.publish_scope for m in middlewares),
                    )

                # Return data for tests
                return result_msg

        # Suitable handler was not found or
        # parsing/decoding exception occurred
        for m in middlewares:
            stack.push_async_exit(m.__aexit__)

        if parsing_error:
            raise parsing_error

        else:
            raise SubscriberNotFound(f"There is no suitable handler for {msg=}")

    # An error was raised and processed by some middleware
    return ensure_response(None)

@override
async def get_one(
    self,
    *,
    timeout: float = 5.0,
    no_ack: bool = True,
) -> "Optional[RabbitMessage]":
    assert self._queue_obj, "You should start subscriber at first."  # nosec B101
    assert (  # nosec B101
        not self.calls
    ), "You can't use `get_one` method if subscriber has registered handlers."

    sleep_interval = timeout / 10

    raw_message: Optional[IncomingMessage] = None
    with anyio.move_on_after(timeout):
        while (  # noqa: ASYNC110
            raw_message := await self._queue_obj.get(
                fail=False,
                no_ack=no_ack,
                timeout=timeout,
            )
        ) is None:
            await anyio.sleep(sleep_interval)

    msg: Optional[RabbitMessage] = await process_msg(  # type: ignore[assignment]
        msg=raw_message,
        middlewares=self._broker_middlewares,
        parser=self._parser,
        decoder=self._decoder,
    )
    return msg

def add_call(
    self,
    *,
    filter_: "Filter[Any]",
    parser_: Optional["CustomCallable"],
    decoder_: Optional["CustomCallable"],
    middlewares_: Iterable["SubscriberMiddleware[Any]"],
    dependencies_: Iterable["Depends"],
) -> Self:
    self._call_options = _CallOptions(
        filter=filter_,
        parser=parser_,
        decoder=decoder_,
        middlewares=middlewares_,
        dependencies=dependencies_,
    )
    return self

def get_description(self) -> Optional[str]:
    """Returns the description of the handler."""
    if not self.calls:  # pragma: no cover
        return None

    else:
        return self.calls[0].description

def get_payloads(self) -> List[Tuple["AnyDict", str]]:
    """Get the payloads of the handler."""
    payloads: List[Tuple[AnyDict, str]] = []

    for h in self.calls:
        if h.dependant is None:
            raise SetupError("You should setup `Handler` at first.")

        body = parse_handler_params(
            h.dependant,
            prefix=f"{self.title_ or self.call_name}:Message",
        )

        payloads.append((body, to_camelcase(h.call_name)))

    if not self.calls:
        payloads.append(
            (
                {
                    "title": f"{self.title_ or self.call_name}:Message:Payload",
                },
                to_camelcase(self.call_name),
            )
        )

    return payloads

@staticmethod
def get_routing_hash(
    queue: "RabbitQueue",
    exchange: Optional["RabbitExchange"] = None,
) -> int:
    """Calculate the routing hash for a RabbitMQ queue and exchange."""
    return hash(queue) + hash(exchange or "")

@staticmethod
def build_log_context(
    message: Optional["StreamMessage[Any]"],
    queue: "RabbitQueue",
    exchange: Optional["RabbitExchange"] = None,
) -> Dict[str, str]:
    return {
        "queue": queue.name,
        "exchange": getattr(exchange, "name", ""),
        "message_id": getattr(message, "message_id", ""),
    }

def get_name(self) -> str:
    return f"{self.queue.name}:{getattr(self.exchange, 'name', None) or '_'}:{self.call_name}"

def get_schema(self) -> Dict[str, Channel]:
    payloads = self.get_payloads()

    return {
        self.name: Channel(
            description=self.description,
            subscribe=Operation(
                bindings=OperationBinding(
                    amqp=amqp.OperationBinding(
                        cc=self.queue.routing,
                    ),
                )
                if is_routing_exchange(self.exchange)
                else None,
                message=Message(
                    title=f"{self.name}:Message",
                    payload=resolve_payloads(payloads),
                    correlationId=CorrelationId(
                        location="$message.header#/correlation_id"
                    ),
                ),
            ),
            bindings=ChannelBinding(
                amqp=amqp.ChannelBinding(
                    **{
                        "is": "routingKey",
                        "queue": amqp.Queue(
                            name=self.queue.name,
                            durable=self.queue.durable,
                            exclusive=self.queue.exclusive,
                            autoDelete=self.queue.auto_delete,
                            vhost=self.virtual_host,
                        )
                        if is_routing_exchange(self.exchange) and self.queue.name
                        else None,
                        "exchange": (
                            amqp.Exchange(type="default", vhost=self.virtual_host)
                            if not self.exchange.name
                            else amqp.Exchange(
                                type=self.exchange.type.value,
                                name=self.exchange.name,
                                durable=self.exchange.durable,
                                autoDelete=self.exchange.auto_delete,
                                vhost=self.virtual_host,
                            )
                        ),
                    }
                )
            ),
        )
    }