Skip to content

RedisStreamMessage

faststream.redis.message.RedisStreamMessage #

RedisStreamMessage(
    *args: Any,
    consumer: ConsumerProtocol,
    group: str | None,
    **kwargs: Any,
)

Bases: _RedisStreamMessageMixin[DefaultStreamMessage]

Source code in faststream/redis/message.py 
167
168
169
170
171
172
def __init__(
    self, *args: Any, consumer: ConsumerProtocol, group: str | None, **kwargs: Any
) -> None:
    super().__init__(*args, **kwargs)
    self.consumer = consumer
    self.group = group

raw_message instance-attribute #

raw_message = raw_message

body instance-attribute #

body = body

reply_to instance-attribute #

reply_to = reply_to

content_type instance-attribute #

content_type = content_type

source_type instance-attribute #

source_type = source_type

headers instance-attribute #

headers = headers or {}

batch_headers instance-attribute #

batch_headers = batch_headers or []

path instance-attribute #

path = path or {}

correlation_id instance-attribute #

correlation_id = correlation_id or str(uuid4())

message_id instance-attribute #

message_id = message_id or self.correlation_id

committed instance-attribute #

committed: AckStatus | None = None

processed instance-attribute #

processed = False

consumer instance-attribute #

consumer = consumer

group instance-attribute #

group = group

get_delivery_count async #

get_delivery_count(
    redis: _DeprecatedType[Redis[bytes]] = EMPTY,
    group: _DeprecatedType[str] = EMPTY,
) -> int

Return this message's current delivery count from the Redis PEL.

The count is queried on every call. Messages without an ID or a pending entry, including acknowledged messages, return 1.

Source code in faststream/redis/message.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
async def get_delivery_count(
    self,
    redis: _DeprecatedType["Redis[bytes]"] = EMPTY,
    group: _DeprecatedType[str] = EMPTY,
) -> int:
    """Return this message's current delivery count from the Redis PEL.

    The count is queried on every call. Messages without an ID or a pending
    entry, including acknowledged messages, return ``1``.
    """
    redis_resolved, group_resolved = self._resolve_consumer_context(redis, group)
    message_ids = self.raw_message["message_ids"]
    if not message_ids or redis_resolved is None or group_resolved is None:
        return 1

    message_id = message_ids[0]
    entries = await redis_resolved.xpending_range(
        name=self.raw_message["channel"],
        groupname=group_resolved,
        min=message_id,
        max=message_id,
        count=1,
    )
    return int(entries[0]["times_delivered"]) if entries else 1

set_decoder #

set_decoder(decoder: AsyncCallable) -> None
Source code in faststream/message/message.py 
69
70
def set_decoder(self, decoder: "AsyncCallable") -> None:
    self.__decoder = decoder

clear_cache #

clear_cache() -> None
Source code in faststream/message/message.py 
72
73
def clear_cache(self) -> None:
    self.__decoded_caches.clear()

decode async #

decode() -> Optional[Any]

Serialize the message by lazy decoder.

Returns a cache after first usage. To prevent such behavior, please call message.clear_cache() after message.body changes.

Source code in faststream/message/message.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
async def decode(self) -> Optional["Any"]:
    """Serialize the message by lazy decoder.

    Returns a cache after first usage. To prevent such behavior, please call
    `message.clear_cache()` after `message.body` changes.
    """
    assert self.__decoder, "You should call `set_decoder()` method first."

    if (
        result := self.__decoded_caches.get(self.__decoder, _NOT_CACHED)
    ) is _NOT_CACHED:
        result = self.__decoded_caches[self.__decoder] = await self.__decoder(self)

    return result

ack async #

ack(
    redis: _DeprecatedType[Optional[Redis[bytes]]] = EMPTY,
    group: _DeprecatedType[str | None] = EMPTY,
) -> None
Source code in faststream/redis/message.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
@override
async def ack(
    self,
    redis: _DeprecatedType[Optional["Redis[bytes]"]] = EMPTY,
    group: _DeprecatedType[str | None] = EMPTY,
) -> None:
    redis_resolved, group_resolved = self._resolve_consumer_context(redis, group)
    if (
        not self.committed
        and group_resolved is not None
        and redis_resolved is not None
    ):
        ids = self.raw_message["message_ids"]
        channel = self.raw_message["channel"]
        await redis_resolved.xack(channel, group_resolved, *ids)
    await super().ack()

nack async #

nack(
    redis: _DeprecatedType[Optional[Redis[bytes]]] = EMPTY,
    group: _DeprecatedType[str | None] = EMPTY,
) -> None
Source code in faststream/redis/message.py 
205
206
207
208
209
210
211
212
@override
async def nack(
    self,
    redis: _DeprecatedType[Optional["Redis[bytes]"]] = EMPTY,
    group: _DeprecatedType[str | None] = EMPTY,
) -> None:
    self._resolve_consumer_context(redis, group)
    await super().nack()

reject async #

reject(
    redis: _DeprecatedType[Optional[Redis[bytes]]] = EMPTY,
    group: _DeprecatedType[str | None] = EMPTY,
) -> None
Source code in faststream/redis/message.py 
214
215
216
217
218
219
220
221
@override
async def reject(
    self,
    redis: _DeprecatedType[Optional["Redis[bytes]"]] = EMPTY,
    group: _DeprecatedType[str | None] = EMPTY,
) -> None:
    self._resolve_consumer_context(redis, group)
    await super().reject()

delete async #

delete(
    redis: _DeprecatedType[Optional[Redis[bytes]]] = EMPTY,
) -> None
Source code in faststream/redis/message.py 
223
224
225
226
227
228
229
230
async def delete(
    self, redis: _DeprecatedType[Optional["Redis[bytes]"]] = EMPTY
) -> None:
    redis_resolved, _ = self._resolve_consumer_context(redis, EMPTY)
    if redis_resolved is not None:
        ids = self.raw_message["message_ids"]
        channel = self.raw_message["channel"]
        await redis_resolved.xdel(channel, *ids)