RabbitSubscriber

def add_call(
    self,
    *,
    parser_: Optional["CustomCallable"],
    decoder_: Optional["CustomCallable"],
    dependencies_: Sequence["Dependant"],
    codec_: Optional["CodecProto"] = None,
) -> Self:
    self._call_options = _CallOptions(
        parser=parser_,
        decoder=decoder_,
        dependencies=dependencies_,
        codec=codec_,
    )
    return self

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.stop()

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

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

    except Exception:  # nosec B110  # ruff: ignore[try-except-pass]
        # All other exceptions were logged by CriticalLogMiddleware
        pass

async def process_message(self, msg: MsgType) -> "Response":
    """Execute all message processing stages."""
    context = self._outer_config.context
    logger_state = self._outer_config.logger

    async with AsyncExitStack() as stack:
        stack.enter_context(self.lock)

        # Enter context before middlewares
        stack.enter_context(
            context.scopes(
                (
                    ("handler_", self),
                    ("logger", logger_state.logger.logger),
                    *self._outer_config.extra_context.items(),
                ),
            ),
        )

        # reserve place for context scope under the middlewares __aexit__
        # because middlewares should be exited before context scope release
        message_scope = stack.enter_context(ExitStack())

        # enter all middlewares
        middlewares: list[BaseMiddleware] = []
        for base_m in self.__build__middlewares_stack():
            middleware = base_m(msg, context=context)
            middlewares.append(middleware)
            await stack.enter_async_context(middleware)

        cache: dict[Any, Any] = {}
        parsing_error: Exception | None = 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:
                message_scope.enter_context(
                    context.scopes(
                        (
                            ("log_context", self.get_log_context(message)),
                            ("message", message),
                        ),
                    ),
                )

                result_msg = ensure_response(
                    await h.call(
                        message=message,
                        # consumer middlewares
                        _extra_middlewares=(
                            m.consume_scope for m in middlewares[::-1]
                        ),
                    ),
                )

                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.as_publish_command(),
                        _extra_middlewares=(
                            m.publish_scope for m in middlewares[::-1]
                        ),
                    )

                # Return data for tests
                return result_msg

        # Suitable handler was not found or
        # parsing/decoding exception occurred.
        # Reraise it to catch in tests
        if parsing_error:
            raise parsing_error

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

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

def schema(self) -> dict[str, "SubscriberSpec"]:
    self._build_fastdepends_model()
    return self.specification.get_schema()

def routing(self) -> str:
    return f"{self._outer_config.prefix}{self.queue.routing()}"

@override
async def start(self) -> None:
    """Starts the consumer for the RabbitMQ queue."""
    await super().start()

    queue_to_bind = self.queue.add_prefix(self._outer_config.prefix)

    declarer = self._outer_config.declarer

    self._queue_obj = queue = await declarer.declare_queue(
        queue_to_bind,
        channel=self.channel,
    )

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

        await queue.bind(
            exchange,
            routing_key=queue_to_bind.routing(),
            arguments=queue_to_bind.bind_arguments,
            timeout=queue_to_bind.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]
            no_ack=self.__no_ack,
            arguments=self.consume_args,
        )

    self._post_start()

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

    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

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

    sleep_interval = timeout / 10

    raw_message: IncomingMessage | None = None
    with (
        contextlib.suppress(asyncio.exceptions.CancelledError),
        anyio.move_on_after(timeout),
    ):
        while (  # ruff: ignore[async-busy-wait]
            raw_message := await self._queue_obj.get(
                fail=False,
                no_ack=no_ack,
                timeout=timeout,
            )
        ) is None:
            await anyio.sleep(sleep_interval)

    context = self._outer_config.context
    async_parser, async_decoder = self._get_parser_and_decoder()

    msg: RabbitMessage | None = await process_msg(  # type: ignore[assignment]
        msg=raw_message,
        middlewares=(
            m(raw_message, context=context) for m in self._broker_middlewares
        ),
        parser=async_parser,
        decoder=async_decoder,
    )
    return msg

@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_log_context(
    self,
    message: Optional["StreamMessage[Any]"],
) -> dict[str, str]:
    return self.build_log_context(
        message=message,
        queue=self.queue,
        exchange=self.exchange,
    )