DefaultPublisher

@override
async def publish(
    self,
    message: "SendableMessage",
    topic: str = "",
    *,
    key: bytes | Any | None = None,
    partition: int | None = None,
    timestamp_ms: int | None = None,
    headers: dict[str, str] | None = None,
    correlation_id: str | None = None,
    reply_to: str = "",
    no_confirm: bool = False,
) -> Union["asyncio.Future[RecordMetadata]", "RecordMetadata"]:
    """Publishes a message to Kafka.

    Args:
        message:
            Message body to send.
        topic:
            Topic where the message will be published.
        key:
            A key to associate with the message. Can be used to
            determine which partition to send the message to. If partition
            is `None` (and producer's partitioner config is left as default),
            then messages with the same key will be delivered to the same
            partition (but if key is `None`, partition is chosen randomly).
            Must be type `bytes`, or be serializable to bytes via configured
            `key_serializer`
        partition:
            Specify a partition. If not set, the partition will be
            selected using the configured `partitioner`
        timestamp_ms:
            Epoch milliseconds (from Jan 1 1970 UTC) to use as
            the message timestamp. Defaults to current time.
        headers:
            Message headers to store metainformation.
        correlation_id:
            Manual message **correlation_id** setter.
            **correlation_id** is a useful option to trace messages.
        reply_to:
            Reply message topic name to send response.
        no_confirm:
            Do not wait for Kafka publish confirmation.

    Returns:
        `asyncio.Future[RecordMetadata]` if no_confirm = True.
        `RecordMetadata` if no_confirm = False.
    """
    cmd = KafkaPublishCommand(
        message,
        topic=topic or self.topic,
        key=key or self.key,
        partition=partition if partition is not None else self.partition,
        reply_to=reply_to or self.reply_to,
        headers=self.headers | (headers or {}),
        correlation_id=correlation_id or self._outer_config.id_generator(),
        timestamp_ms=timestamp_ms,
        no_confirm=no_confirm,
        _publish_type=PublishType.PUBLISH,
    )
    return await self._basic_publish(
        cmd,
        producer=self._outer_config.producer,
        _extra_middlewares=(),
    )

@override
async def request(
    self,
    message: "SendableMessage",
    topic: str = "",
    *,
    key: bytes | Any | None = None,
    partition: int | None = None,
    timestamp_ms: int | None = None,
    headers: dict[str, str] | None = None,
    correlation_id: str | None = None,
    timeout: float = 0.5,
) -> "KafkaMessage":
    """Send a request message and wait for a response.

    Args:
        message: Message body to send.
        topic: Topic where the message will be published.
        key: A key to associate with the message. Can be used to
            determine which partition to send the message to. If partition
            is `None` (and producer's partitioner config is left as default),
            then messages with the same key will be delivered to the same
            partition (but if key is `None`, partition is chosen randomly).
            Must be type `bytes`, or be serializable to bytes via configured
            `key_serializer`.
        partition: Specify a partition. If not set, the partition will be
            selected using the configured `partitioner`.
        timestamp_ms: Epoch milliseconds (from Jan 1 1970 UTC) to use as
            the message timestamp. Defaults to current time.
        headers: Message headers to store metainformation.
        correlation_id: Manual message **correlation_id** setter.
            **correlation_id** is a useful option to trace messages.
        timeout: Timeout to send RPC request.

    Returns:
        The response message.
    """
    return await super().request(
        message,
        topic=topic,
        key=key or self.key,
        partition=partition,
        timestamp_ms=timestamp_ms,
        headers=headers,
        correlation_id=correlation_id,
        timeout=timeout,
    )

async def assert_called_once_with(
    self,
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
) -> None:
    """Assert the endpoint was called once, with the message described here.

    Args:
        body: The body as a dict, a model or a matcher; it goes through the codec.
        headers: Headers the message must carry; the rest may carry more.
        correlation_id: The exact correlation id.
        reply_to: The exact reply-to destination.
        content_type: The exact content type.
        path: The exact path parameters the subject template matched.
        context: Context paths, as given to `Context()`, mapped to their values.
    """
    recorder = self._recorder_with_calls()
    recorder.mock.assert_called_once()
    await recorder.assert_last_call(
        ExpectedCall(
            body=body,
            headers=headers,
            correlation_id=correlation_id,
            reply_to=reply_to,
            content_type=content_type,
            path=path,
            context=context,
        )
    )

async def assert_called_with(
    self,
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
) -> None:
    """Assert the last message the endpoint saw is the one described here.

    Args:
        body: The body as a dict, a model or a matcher; it goes through the codec.
        headers: Headers the message must carry; the rest may carry more.
        correlation_id: The exact correlation id.
        reply_to: The exact reply-to destination.
        content_type: The exact content type.
        path: The exact path parameters the subject template matched.
        context: Context paths, as given to `Context()`, mapped to their values.
    """
    recorder = self._recorder_with_calls()
    await recorder.assert_last_call(
        ExpectedCall(
            body=body,
            headers=headers,
            correlation_id=correlation_id,
            reply_to=reply_to,
            content_type=content_type,
            path=path,
            context=context,
        )
    )

async def assert_any_call(
    self,
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
) -> None:
    """Assert one of the messages the endpoint saw is the one described here.

    Args:
        body: The body as a dict, a model or a matcher; it goes through the codec.
        headers: Headers the message must carry; the rest may carry more.
        correlation_id: The exact correlation id.
        reply_to: The exact reply-to destination.
        content_type: The exact content type.
        path: The exact path parameters the subject template matched.
        context: Context paths, as given to `Context()`, mapped to their values.
    """
    recorder = self._recorder_with_calls()
    await recorder.assert_any_call(
        ExpectedCall(
            body=body,
            headers=headers,
            correlation_id=correlation_id,
            reply_to=reply_to,
            content_type=content_type,
            path=path,
            context=context,
        )
    )

async def start(self) -> None:
    pass

def set_test(
    self,
    *,
    recorder: CallRecorder | None = None,
    with_fake: bool,
) -> None:
    """Turn publisher to testing mode, sharing `recorder` when one is given."""
    self.is_test = True
    if recorder is None:
        self._recorder.reset()
    else:
        self._recorder = recorder
    self._fake_handler = with_fake

def reset_test(self) -> None:
    """Turn off publisher's testing mode."""
    self.is_test = False
    self._recorder.reset()
    # A shared recorder goes back to the handler it belongs to
    self._recorder = CallRecorder(self.specification.name, self._outer_config)
    self._fake_handler = False

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

async def flush(self) -> None:
    producer = cast("AioKafkaFastProducer", self._outer_config.producer)
    await producer.flush()