Skip to content

MessageFormat

faststream.redis.parser.MessageFormat #

MessageFormat(
    data: bytes, headers: dict[str, Any] | None = None
)

Bases: ABC

A class to represent a raw Redis message.

Source code in faststream/redis/parser/message.py 
21
22
23
24
25
26
27
def __init__(
    self,
    data: bytes,
    headers: dict[str, Any] | None = None,
) -> None:
    self.data = data
    self.headers = headers or {}

data instance-attribute #

data = data

headers instance-attribute #

headers = headers or {}

build classmethod #

build(
    *,
    message: Union[
        Sequence[SendableMessage], SendableMessage
    ],
    reply_to: str | None,
    headers: dict[str, Any] | None,
    correlation_id: str,
    serializer: Optional[SerializerProto] = None,
) -> MessageFormat
Source code in faststream/redis/parser/message.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@classmethod
def build(
    cls,
    *,
    message: Union[Sequence["SendableMessage"], "SendableMessage"],
    reply_to: str | None,
    headers: dict[str, Any] | None,
    correlation_id: str,
    serializer: Optional["SerializerProto"] = None,
) -> "MessageFormat":
    payload, content_type = encode_message(message, serializer=serializer)

    headers_to_send = {
        "correlation_id": correlation_id,
    }

    if content_type:
        headers_to_send["content-type"] = content_type

    if reply_to:
        headers_to_send["reply_to"] = reply_to

    if headers is not None:
        headers_to_send.update(headers)

    return cls(
        data=payload,
        headers=headers_to_send,
    )

encode abstractmethod classmethod #

encode(
    *,
    message: Union[
        Sequence[SendableMessage], SendableMessage
    ],
    reply_to: str | None,
    headers: dict[str, Any] | None,
    correlation_id: str,
    serializer: Optional[SerializerProto] = None,
) -> bytes
Source code in faststream/redis/parser/message.py 
59
60
61
62
63
64
65
66
67
68
69
70
@classmethod
@abstractmethod
def encode(
    cls,
    *,
    message: Union[Sequence["SendableMessage"], "SendableMessage"],
    reply_to: str | None,
    headers: dict[str, Any] | None,
    correlation_id: str,
    serializer: Optional["SerializerProto"] = None,
) -> bytes:
    raise NotImplementedError

parse abstractmethod classmethod #

parse(data: bytes) -> tuple[bytes, dict[str, Any]]
Source code in faststream/redis/parser/message.py 
72
73
74
75
@classmethod
@abstractmethod
def parse(cls, data: bytes) -> tuple[bytes, dict[str, Any]]:
    raise NotImplementedError