ZmqttProducerV5

@override
async def publish(self, cmd: "MQTTPublishCommand") -> None:
    payload, content_type = await self.codec.encode(cmd.body, self.serializer)

    user_props: list[tuple[str, str]] = [
        (k, str(v)) for k, v in (cmd.headers or {}).items()
    ]

    properties = PublishProperties(
        content_type=content_type or None,
        response_topic=cmd.reply_to or None,
        correlation_data=cmd.correlation_id.encode() if cmd.correlation_id else None,
        user_properties=tuple(user_props),
        message_expiry_interval=cmd.message_expiry_interval,
    )

    await self._connected_client.publish(
        cmd.destination,
        payload,
        qos=cmd.qos,
        retain=cmd.retain,
        properties=properties,
    )

@override
async def request(self, cmd: "MQTTPublishCommand") -> "zmqtt.Message":
    """Request/reply for MQTT 5.0 via zmqtt's native client.request().

    zmqtt auto-generates a unique reply topic.  We pass our correlation
    ID explicitly so the responder echoes it back and the caller can
    verify it on the response StreamMessage.
    """
    payload, content_type = await self.codec.encode(cmd.body, self.serializer)
    correlation_id = cmd.correlation_id or self.id_generator()

    user_props: list[tuple[str, str]] = [
        (k, str(v)) for k, v in (cmd.headers or {}).items()
    ]

    # Pass correlation_data explicitly so the responder echoes it back.
    # Do NOT set response_topic — let zmqtt generate it.
    properties = PublishProperties(
        content_type=content_type or None,
        correlation_data=correlation_id.encode(),
        user_properties=tuple(user_props),
        message_expiry_interval=cmd.message_expiry_interval,
    )

    return await self._connected_client.request(
        cmd.destination,
        payload,
        qos=cmd.qos,
        timeout=cmd.timeout or 30.0,
        properties=properties,
    )

@override
async def publish_batch(self, cmd: "MQTTPublishCommand") -> None:
    msg = "MQTT does not support batch publishing."
    raise FeatureNotSupportedException(msg)

def connect(
    self,
    client: "zmqtt.MQTTClient",
    serializer: Optional["SerializerProto"],
    codec: Optional["CodecProto"] = None,
) -> None:
    self._client = client
    self.serializer = serializer
    self.codec = codec or DefaultCodec()

def disconnect(self) -> None:
    self._client = None
    self.serializer = None