BasePrometheusMiddleware

async def consume_scope(
    self,
    call_next: "AsyncFuncAny",
    msg: "StreamMessage[AnyMsg]",
) -> Any:
    if self._settings_provider is None or msg.source_type is SourceType.RESPONSE:
        return await call_next(msg)

    messaging_system = self._settings_provider.messaging_system
    consume_attrs = self._settings_provider.get_consume_attrs_from_message(msg)
    destination_name = consume_attrs["destination_name"]

    self._metrics_manager.add_received_message(
        amount=consume_attrs["messages_count"],
        broker=messaging_system,
        handler=destination_name,
        **self._custom_labels,
    )

    self._metrics_manager.observe_received_messages_size(
        size=consume_attrs["message_size"],
        broker=messaging_system,
        handler=destination_name,
        **self._custom_labels,
    )

    self._metrics_manager.add_received_message_in_process(
        amount=consume_attrs["messages_count"],
        broker=messaging_system,
        handler=destination_name,
        **self._custom_labels,
    )

    err: Exception | None = None
    start_time = time.perf_counter()

    try:
        result = await call_next(msg)

    except Exception as e:
        err = e

        if not isinstance(err, IgnoredException):
            self._metrics_manager.add_received_processed_message_exception(
                exception_type=type(err).__name__,
                broker=messaging_system,
                handler=destination_name,
                **self._custom_labels,
            )
        raise

    finally:
        duration = time.perf_counter() - start_time
        self._metrics_manager.observe_received_processed_message_duration(
            duration=duration,
            broker=messaging_system,
            handler=destination_name,
            **self._custom_labels,
        )

        self._metrics_manager.remove_received_message_in_process(
            amount=consume_attrs["messages_count"],
            broker=messaging_system,
            handler=destination_name,
            **self._custom_labels,
        )

        status = ProcessingStatus.acked

        if msg.committed or err:
            status = (
                PROCESSING_STATUS_BY_ACK_STATUS.get(msg.committed)  # type: ignore[arg-type]
                or PROCESSING_STATUS_BY_HANDLER_EXCEPTION_MAP.get(type(err))
                or ProcessingStatus.error
            )

        self._metrics_manager.add_received_processed_message(
            amount=consume_attrs["messages_count"],
            status=status,
            broker=messaging_system,
            handler=destination_name,
            **self._custom_labels,
        )

    return result

async def publish_scope(
    self,
    call_next: Callable[[PublishCommandType], Awaitable[Any]],
    cmd: PublishCommandType,
) -> Any:
    if self._settings_provider is None or cmd.publish_type is PublishType.REPLY:
        return await call_next(cmd)

    destination_name = self._settings_provider.get_publish_destination_name_from_cmd(
        cmd
    )
    messaging_system = self._settings_provider.messaging_system

    err: Exception | None = None
    start_time = time.perf_counter()

    try:
        result = await call_next(cmd)

    except Exception as e:
        err = e
        self._metrics_manager.add_published_message_exception(
            exception_type=type(err).__name__,
            broker=messaging_system,
            destination=destination_name,
            **self._custom_labels,
        )
        raise

    finally:
        duration = time.perf_counter() - start_time

        self._metrics_manager.observe_published_message_duration(
            duration=duration,
            broker=messaging_system,
            destination=destination_name,
            **self._custom_labels,
        )

        status = PublishingStatus.error if err else PublishingStatus.success

        self._metrics_manager.add_published_message(
            amount=len(cmd.batch_bodies),
            status=status,
            broker=messaging_system,
            destination=destination_name,
            **self._custom_labels,
        )

    return result

async def on_receive(self) -> None:
    """Hook to call on message receive."""

async def after_processed(  # ruff: ignore[no-self-use]
    self,
    exc_type: type[BaseException] | None = None,
    exc_val: BaseException | None = None,
    exc_tb: Optional["TracebackType"] = None,
) -> bool | None:
    """Asynchronously called after processing."""
    return False