Skip to content

RedisBatchStreamMessage

faststream.redis.message.RedisBatchStreamMessage dataclass #

RedisBatchStreamMessage(
    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(),
    _source_type=Consume,
)

Bases: _RedisStreamMessageMixin[BatchStreamMessage]

decoded_body instance-attribute #

decoded_body

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)

ack async #

ack(redis=None, group=None)
Source code in faststream/redis/message.py 
121
122
123
124
125
126
127
128
129
130
131
@override
async def ack(
    self,
    redis: Optional["Redis[bytes]"] = None,
    group: Optional[str] = None,
) -> None:
    if not self.committed and group is not None and redis is not None:
        ids = self.raw_message["message_ids"]
        channel = self.raw_message["channel"]
        await redis.xack(channel, group, *ids)  # type: ignore[no-untyped-call]
    await super().ack()

nack async #

nack(redis=None, group=None)
Source code in faststream/redis/message.py 
133
134
135
136
137
138
139
@override
async def nack(
    self,
    redis: Optional["Redis[bytes]"] = None,
    group: Optional[str] = None,
) -> None:
    await super().nack()

reject async #

reject(redis=None, group=None)
Source code in faststream/redis/message.py 
141
142
143
144
145
146
147
@override
async def reject(
    self,
    redis: Optional["Redis[bytes]"] = None,
    group: Optional[str] = None,
) -> None:
    await super().reject()

decode async #

decode()

Serialize the message by lazy decoder.

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

delete async #

delete(redis)
Source code in faststream/redis/message.py 
149
150
151
152
153
async def delete(self, redis: Optional["Redis[bytes]"]) -> None:
    if redis is not None:
        ids = self.raw_message["message_ids"]
        channel = self.raw_message["channel"]
        await redis.xdel(channel, *ids)