BaseTelemetryMiddleware

async def publish_scope(
    self,
    call_next: "AsyncFunc",
    msg: "PublishCommandType",
) -> Any:
    if (provider := self.__settings_provider) is None:
        return await call_next(msg)

    headers = msg.headers
    current_context = context.get_current()
    destination_name = provider.get_publish_destination_name(msg)

    current_baggage: Baggage | None = self.context.get_local("baggage")
    if current_baggage:
        headers.update(current_baggage.to_headers())

    trace_attributes = provider.get_publish_attrs_from_cmd(msg)
    metrics_attributes = {
        SpanAttributes.MESSAGING_SYSTEM: provider.messaging_system,
        SpanAttributes.MESSAGING_DESTINATION_NAME: destination_name,
    }

    # NOTE: if batch with single message?
    if (msg_count := len(msg.batch_bodies)) > 1:
        trace_attributes[SpanAttributes.MESSAGING_BATCH_MESSAGE_COUNT] = msg_count
        current_context = _BAGGAGE_PROPAGATOR.extract(headers, current_context)
        _BAGGAGE_PROPAGATOR.inject(
            headers,
            baggage.set_baggage(WITH_BATCH, True, context=current_context),
        )

    if self._current_span and self._current_span.is_recording():
        current_context = trace.set_span_in_context(
            self._current_span,
            current_context,
        )
        _TRACE_PROPAGATOR.inject(headers, context=self._origin_context)

    else:
        create_span = self._tracer.start_span(
            name=_create_span_name(destination_name, MessageAction.CREATE),
            kind=trace.SpanKind.PRODUCER,
            attributes=trace_attributes,
        )
        current_context = trace.set_span_in_context(create_span)
        _TRACE_PROPAGATOR.inject(headers, context=current_context)
        create_span.end()

    start_time = time.perf_counter()

    try:
        with self._tracer.start_as_current_span(
            name=_create_span_name(destination_name, MessageAction.PUBLISH),
            kind=trace.SpanKind.PRODUCER,
            attributes=trace_attributes,
            context=current_context,
        ) as span:
            span.set_attribute(
                SpanAttributes.MESSAGING_OPERATION,
                MessageAction.PUBLISH,
            )
            msg.headers = headers
            result = await call_next(msg)

    except Exception as e:
        metrics_attributes[ERROR_TYPE] = type(e).__name__
        raise

    finally:
        duration = time.perf_counter() - start_time
        self._metrics.observe_publish(metrics_attributes, duration, msg_count)

    for key, token in self._scope_tokens:
        self.context.reset_local(key, token)

    return result

async def consume_scope(
    self,
    call_next: "AsyncFuncAny",
    msg: "StreamMessage[Any]",
) -> Any:
    if (provider := self.__settings_provider) is None:
        return await call_next(msg)

    if _is_batch_message(msg):
        links = _get_msg_links(msg)
        current_context = Context()
    else:
        links = None
        current_context = _TRACE_PROPAGATOR.extract(msg.headers)

    destination_name = provider.get_consume_destination_name(msg)
    trace_attributes = provider.get_consume_attrs_from_message(msg)
    metrics_attributes = {
        SpanAttributes.MESSAGING_SYSTEM: provider.messaging_system,
        MESSAGING_DESTINATION_PUBLISH_NAME: destination_name,
    }

    if not len(current_context):
        create_span = self._tracer.start_span(
            name=_create_span_name(destination_name, MessageAction.CREATE),
            kind=trace.SpanKind.CONSUMER,
            attributes=trace_attributes,
            links=links,
        )
        current_context = trace.set_span_in_context(create_span)
        create_span.end()

    self._origin_context = current_context
    start_time = time.perf_counter()

    try:
        with self._tracer.start_as_current_span(
            name=_create_span_name(destination_name, MessageAction.PROCESS),
            kind=trace.SpanKind.CONSUMER,
            context=current_context,
            attributes=trace_attributes,
            end_on_exit=False,
        ) as span:
            span.set_attribute(
                SpanAttributes.MESSAGING_OPERATION,
                MessageAction.PROCESS,
            )
            self._current_span = span

            self._scope_tokens.append((
                "span",
                self.context.set_local("span", span),
            ))
            self._scope_tokens.append(
                (
                    "baggage",
                    self.context.set_local("baggage", Baggage.from_msg(msg)),
                ),
            )

            new_context = trace.set_span_in_context(span, current_context)
            token = context.attach(new_context)
            result = await call_next(msg)
            context.detach(token)

    except Exception as e:
        metrics_attributes[ERROR_TYPE] = type(e).__name__
        raise

    finally:
        duration = time.perf_counter() - start_time
        msg_count = trace_attributes.get(
            SpanAttributes.MESSAGING_BATCH_MESSAGE_COUNT,
            1,
        )
        self._metrics.observe_consume(metrics_attributes, duration, msg_count)

    return result

async def after_processed(
    self,
    exc_type: type[BaseException] | None = None,
    exc_val: BaseException | None = None,
    exc_tb: Optional["TracebackType"] = None,
) -> bool | None:
    if self._current_span and self._current_span.is_recording():
        self._current_span.end()
    return False

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