build_message

faststream.redis.testing.build_message #

build_message(
    message: Union[
        Sequence[SendableMessage], SendableMessage
    ],
    *,
    correlation_id: str,
    message_format: type[MessageFormat],
    reply_to: str = "",
    headers: dict[str, Any] | None = None,
    serializer: Optional[SerializerProto] = None,
) -> bytes

Source code in faststream/redis/testing.py

def build_message(
    message: Union[Sequence["SendableMessage"], "SendableMessage"],
    *,
    correlation_id: str,
    message_format: type["MessageFormat"],
    reply_to: str = "",
    headers: dict[str, Any] | None = None,
    serializer: Optional["SerializerProto"] = None,
) -> bytes:
    return message_format.encode(
        message=message,
        reply_to=reply_to,
        headers=headers,
        correlation_id=correlation_id,
        serializer=serializer,
    )