Skip to content

NatsBatchMessage

faststream.nats.message.NatsBatchMessage dataclass #

NatsBatchMessage(raw_message, body, headers=dict(), batch_headers=list(), path=dict(), content_type=None, reply_to='', message_id=gen_cor_id(), correlation_id=gen_cor_id())

Bases: StreamMessage[List[Msg]]

A class to represent a NATS batch message.

raw_message instance-attribute #

raw_message

body instance-attribute #

body

headers class-attribute instance-attribute #

headers = field(default_factory=dict)

batch_headers class-attribute instance-attribute #

batch_headers = field(default_factory=list)

path class-attribute instance-attribute #

path = field(default_factory=dict)

content_type class-attribute instance-attribute #

content_type = None

reply_to class-attribute instance-attribute #

reply_to = ''

message_id class-attribute instance-attribute #

message_id = field(default_factory=gen_cor_id)

correlation_id class-attribute instance-attribute #

correlation_id = field(default_factory=gen_cor_id)

processed class-attribute instance-attribute #

processed = field(default=False, init=False)

committed class-attribute instance-attribute #

committed = field(default=None, init=False)

decoded_body property writable #

decoded_body

decode async #

decode()

Serialize the message by lazy decoder.

Source code in faststream/broker/message.py 
74
75
76
77
async def decode(self) -> Optional["DecodedMessage"]:
    """Serialize the message by lazy decoder."""
    # TODO: make it lazy after `decoded_body` removed
    return self._decoded_body

ack async #

ack()
Source code in faststream/nats/message.py 
41
42
43
44
45
46
47
48
async def ack(self) -> None:
    for m in filter(
        lambda m: not m._ackd,
        self.raw_message,
    ):
        await m.ack()

    await super().ack()

nack async #

nack(delay=None)
Source code in faststream/nats/message.py 
50
51
52
53
54
55
56
57
58
59
60
async def nack(
    self,
    delay: Union[int, float, None] = None,
) -> None:
    for m in filter(
        lambda m: not m._ackd,
        self.raw_message,
    ):
        await m.nak(delay=delay)

    await super().nack()

reject async #

reject()
Source code in faststream/nats/message.py 
62
63
64
65
66
67
68
69
async def reject(self) -> None:
    for m in filter(
        lambda m: not m._ackd,
        self.raw_message,
    ):
        await m.term()

    await super().reject()

in_progress async #

in_progress()
Source code in faststream/nats/message.py 
71
72
73
74
75
76
async def in_progress(self) -> None:
    for m in filter(
        lambda m: not m._ackd,
        self.raw_message,
    ):
        await m.in_progress()