Handler

def add_call(
    self,
    *,
    handler: HandlerCallWrapper[
        aio_pika.IncomingMessage, P_HandlerParams, T_HandlerReturn
    ],
    dependant: CallModel[P_HandlerParams, T_HandlerReturn],
    parser: Optional[CustomParser[aio_pika.IncomingMessage, RabbitMessage]],
    decoder: Optional[CustomDecoder[RabbitMessage]],
    filter: Filter[RabbitMessage],
    middlewares: Optional[
        Sequence[Callable[[aio_pika.IncomingMessage], BaseMiddleware]]
    ],
) -> None:
    """Add a call to the handler.

    Args:
        handler: The handler for the call.
        dependant: The dependant for the call.
        parser: Optional custom parser for the call.
        decoder: Optional custom decoder for the call.
        filter: The filter for the call.
        middlewares: Optional sequence of middlewares for the call.

    Returns:
        None

    """
    super().add_call(
        handler=handler,
        parser=resolve_custom_func(parser, AioPikaParser.parse_message),
        decoder=resolve_custom_func(decoder, AioPikaParser.decode_message),
        filter=filter,  # type: ignore[arg-type]
        dependant=dependant,
        middlewares=middlewares,
    )

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
            await self._queue_obj.cancel(self._consumer_tag)
            self._consumer_tag = None
        self._queue_obj = None

@override
async def consume(self, msg: MsgType) -> SendableMessage:  # type: ignore[override]
    """Consume a message asynchronously.

    Args:
        msg: The message to be consumed.

    Returns:
        The sendable message.

    Raises:
        StopConsume: If the consumption needs to be stopped.

    Raises:
        Exception: If an error occurs during consumption.

    """
    result: Optional[WrappedReturn[SendableMessage]] = None
    result_msg: SendableMessage = None

    if not self.running:
        return result_msg

    log_context_tag: Optional["Token[Any]"] = None
    async with AsyncExitStack() as stack:
        stack.enter_context(self.lock)

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

        gl_middlewares: List[BaseMiddleware] = [
            await stack.enter_async_context(m(msg)) for m in self.global_middlewares
        ]

        logged = False
        processed = False
        for handler, filter_, parser, decoder, middlewares, _ in self.calls:
            local_middlewares: List[BaseMiddleware] = [
                await stack.enter_async_context(m(msg)) for m in middlewares
            ]

            all_middlewares = gl_middlewares + local_middlewares

            # TODO: add parser & decoder caches
            message = await parser(msg)

            if not logged:  # pragma: no branch
                log_context_tag = context.set_local(
                    "log_context",
                    self.log_context_builder(message),
                )

            message.decoded_body = await decoder(message)
            message.processed = processed

            if await filter_(message):
                assert (  # nosec B101
                    not processed
                ), "You can't process a message with multiple consumers"

                try:
                    async with AsyncExitStack() as consume_stack:
                        for m_consume in all_middlewares:
                            message.decoded_body = (
                                await consume_stack.enter_async_context(
                                    m_consume.consume_scope(message.decoded_body)
                                )
                            )

                        result = await cast(
                            Awaitable[Optional[WrappedReturn[SendableMessage]]],
                            handler.call_wrapped(message),
                        )

                    if result is not None:
                        result_msg, pub_response = result

                        # TODO: suppress all publishing errors and raise them after all publishers will be tried
                        for publisher in (pub_response, *handler._publishers):
                            if publisher is not None:
                                async with AsyncExitStack() as pub_stack:
                                    result_to_send = result_msg

                                    for m_pub in all_middlewares:
                                        result_to_send = (
                                            await pub_stack.enter_async_context(
                                                m_pub.publish_scope(result_to_send)
                                            )
                                        )

                                    await publisher.publish(
                                        message=result_to_send,
                                        correlation_id=message.correlation_id,
                                    )

                except StopConsume:
                    await self.close()
                    handler.trigger()

                except HandlerException as e:  # pragma: no cover
                    handler.trigger()
                    raise e

                except Exception as e:
                    handler.trigger(error=e)
                    raise e

                else:
                    handler.trigger(result=result[0] if result else None)
                    message.processed = processed = True
                    if IS_OPTIMIZED:  # pragma: no cover
                        break

        assert not self.running or processed, "You have to consume message"  # nosec B101

    if log_context_tag is not None:
        context.reset_local("log_context", log_context_tag)

    return result_msg

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

    for h, _, _, _, _, dep in self.calls:
        body = parse_handler_params(
            dep, prefix=f"{self._title or self.call_name}:Message"
        )
        payloads.append((body, to_camelcase(unwrap(h._original_call).__name__)))

    return payloads

@abstractproperty
def name(self) -> str:
    """Returns the name of the API operation."""
    raise NotImplementedError()

def schema(self) -> Dict[str, Channel]:
    if not self.include_in_schema:
        return {}

    payloads = self.get_payloads()

    handler_name = (
        self._title
        or f"{self.queue.name}:{getattr(self.exchange, 'name', '_')}:{self.call_name}"
    )

    return {
        handler_name: Channel(
            description=self.description,  # type: ignore[attr-defined]
            subscribe=Operation(
                bindings=OperationBinding(
                    amqp=amqp.OperationBinding(
                        cc=self.queue.routing,
                    ),
                )
                if _is_exchange(self.exchange)
                else None,
                message=Message(
                    title=f"{handler_name}:Message",
                    payload=resolve_payloads(payloads),
                    correlationId=CorrelationId(
                        location="$message.header#/correlation_id"
                    ),
                ),
            ),
            bindings=ChannelBinding(
                amqp=amqp.ChannelBinding(
                    **{
                        "is": "routingKey",  # type: ignore
                        "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_exchange(self.exchange)
                        else None,
                        "exchange": (
                            amqp.Exchange(type="default", vhost=self.virtual_host)
                            if self.exchange is None
                            else amqp.Exchange(
                                type=self.exchange.type,  # type: ignore
                                name=self.exchange.name,
                                durable=self.exchange.durable,
                                autoDelete=self.exchange.auto_delete,
                                vhost=self.virtual_host,
                            )
                        ),
                    }
                )
            ),
        )
    }

@override
async def start(self, declarer: RabbitDeclarer) -> None:  # type: ignore[override]
    """Starts the consumer for the RabbitMQ queue.

    Args:
        declarer: RabbitDeclarer object used to declare the queue and exchange

    Returns:
        None

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

    if self.exchange is not None:
        exchange = await declarer.declare_exchange(self.exchange)
        await queue.bind(
            exchange,
            routing_key=self.queue.routing,
            arguments=self.queue.bind_arguments,
        )

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

    await super().start()