Skip to content

RawMessage

faststream.redis.parser.RawMessage #

RawMessage(data, headers=None)

A class to represent a raw Redis message.

Source code in faststream/redis/parser.py 
48
49
50
51
52
53
54
def __init__(
    self,
    data: bytes,
    headers: Optional["AnyDict"] = None,
) -> None:
    self.data = data
    self.headers = headers or {}

data instance-attribute #

data = data

headers instance-attribute #

headers = headers or {}

build classmethod #

build(*, message, reply_to, headers, correlation_id)
Source code in faststream/redis/parser.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@classmethod
def build(
    cls,
    *,
    message: Union[Sequence["SendableMessage"], "SendableMessage"],
    reply_to: Optional[str],
    headers: Optional["AnyDict"],
    correlation_id: str,
) -> "RawMessage":
    payload, content_type = encode_message(message)

    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 classmethod #

encode(*, message, reply_to, headers, correlation_id)
Source code in faststream/redis/parser.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@classmethod
def encode(
    cls,
    *,
    message: Union[Sequence["SendableMessage"], "SendableMessage"],
    reply_to: Optional[str],
    headers: Optional["AnyDict"],
    correlation_id: str,
) -> bytes:
    msg = cls.build(
        message=message,
        reply_to=reply_to,
        headers=headers,
        correlation_id=correlation_id,
    )

    return dump_json(
        {
            "data": msg.data,
            "headers": msg.headers,
        }
    )

parse staticmethod #

parse(data)
Source code in faststream/redis/parser.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
@staticmethod
def parse(data: bytes) -> Tuple[bytes, "AnyDict"]:
    headers: AnyDict

    try:
        # FastStream message format
        parsed_data = json_loads(data)
        data = parsed_data["data"].encode()
        headers = parsed_data["headers"]

    except Exception:
        # Raw Redis message format
        data = data
        headers = {}

    return data, headers