RedisStreamMessage

faststream.redis.message.RedisStreamMessage

RedisStreamMessage(
    raw_message,
    body,
    *,
    headers=None,
    reply_to="",
    batch_headers=None,
    path=None,
    content_type=None,
    correlation_id=None,
    message_id=None,
    source_type=CONSUME,
)

Bases: _RedisStreamMessageMixin[DefaultStreamMessage]

Source code in faststream/message/message.py

def __init__(
    self,
    raw_message: "MsgType",
    body: bytes | Any,
    *,
    headers: dict[str, Any] | None = None,
    reply_to: str = "",
    batch_headers: list[dict[str, Any]] | None = None,
    path: dict[str, Any] | None = None,
    content_type: str | None = None,
    correlation_id: str | None = None,
    message_id: str | None = None,
    source_type: SourceType = SourceType.CONSUME,
) -> None:
    self.raw_message = raw_message
    self.body = body
    self.reply_to = reply_to
    self.content_type = content_type
    self.source_type = source_type

    self.headers = headers or {}
    self.batch_headers = batch_headers or []
    self.path = path or {}
    self.correlation_id = correlation_id or str(uuid4())
    self.message_id = message_id or self.correlation_id

    self.committed: AckStatus | None = None
    self.processed = False

    # Setup later
    self.__decoder: AsyncCallable | None = None
    self.__decoded_caches: dict[
        Any,
        Any,
    ] = {}  # Cache values between filters and tests

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 correlation_id

committed instance-attribute

committed = None

processed instance-attribute

processed = False

set_decoder

set_decoder(decoder)

Source code in faststream/message/message.py

def set_decoder(self, decoder: "AsyncCallable") -> None:
    self.__decoder = decoder

clear_cache

clear_cache()

Source code in faststream/message/message.py

def clear_cache(self) -> None:
    self.__decoded_caches.clear()

decode async

decode()

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

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)) is None:
        result = self.__decoded_caches[self.__decoder] = await self.__decoder(self)

    return result

ack async

ack(redis=None, group=None)

Source code in faststream/redis/message.py

@override
async def ack(
    self,
    redis: Optional["Redis[bytes]"] = None,
    group: str | None = 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

@override
async def nack(
    self,
    redis: Optional["Redis[bytes]"] = None,
    group: str | None = None,
) -> None:
    await super().nack()

reject async

reject(redis=None, group=None)

Source code in faststream/redis/message.py

@override
async def reject(
    self,
    redis: Optional["Redis[bytes]"] = None,
    group: str | None = None,
) -> None:
    await super().reject()

delete async

delete(redis)

Source code in faststream/redis/message.py

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)