Skip to content

build_message

faststream.nats.testing.build_message #

build_message(message, subject, *, reply_to='', correlation_id=None, headers=None)
Source code in faststream/nats/testing.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def build_message(
    message: "SendableMessage",
    subject: str,
    *,
    reply_to: str = "",
    correlation_id: Optional[str] = None,
    headers: Optional[Dict[str, str]] = None,
) -> "PatchedMessage":
    msg, content_type = encode_message(message)
    return PatchedMessage(
        _client=None,  # type: ignore
        subject=subject,
        reply=reply_to,
        data=msg,
        headers={
            "content-type": content_type or "",
            "correlation_id": correlation_id or gen_cor_id(),
            **(headers or {}),
        },
    )