Skip to content

RedisFastProducer

faststream.redis.publisher.producer.RedisFastProducer #

RedisFastProducer(
    connection, parser, decoder, message_format, serializer
)

Bases: ProducerProto[RedisPublishCommand]

A class to represent a Redis producer.

Source code in faststream/redis/publisher/producer.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(
    self,
    connection: "ConnectionState",
    parser: Optional["CustomCallable"],
    decoder: Optional["CustomCallable"],
    message_format: type["MessageFormat"],
    serializer: Optional["SerializerProto"],
) -> None:
    self._connection = connection

    default = RedisPubSubParser(SimpleParserConfig(message_format))
    self._parser = ParserComposition(
        parser,
        default.parse_message,
    )
    self._decoder = ParserComposition(
        decoder,
        default.decode_message,
    )
    self.serializer = serializer

serializer instance-attribute #

serializer = serializer

publish async #

publish(cmd)
Source code in faststream/redis/publisher/producer.py 
49
50
51
52
53
54
55
56
57
58
59
@override
async def publish(self, cmd: "RedisPublishCommand") -> int | bytes:
    msg = cmd.message_format.encode(
        message=cmd.body,
        reply_to=cmd.reply_to,
        headers=cmd.headers,
        correlation_id=cmd.correlation_id or "",
        serializer=self.serializer,
    )

    return await self.__publish(msg, cmd)

request async #

request(cmd)
Source code in faststream/redis/publisher/producer.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@override
async def request(self, cmd: "RedisPublishCommand") -> "Any":
    nuid = NUID()
    reply_to = str(nuid.next(), "utf-8")
    psub = self._connection.client.pubsub()

    try:
        await psub.subscribe(reply_to)

        msg = cmd.message_format.encode(
            message=cmd.body,
            reply_to=reply_to,
            headers=cmd.headers,
            correlation_id=cmd.correlation_id or "",
            serializer=self.serializer,
        )

        await self.__publish(msg, cmd)

        with anyio.fail_after(cmd.timeout) as scope:
            # skip subscribe message
            await psub.get_message(
                ignore_subscribe_messages=True,
                timeout=cmd.timeout or 0.0,
            )

            # get real response
            response_msg = await psub.get_message(
                ignore_subscribe_messages=True,
                timeout=cmd.timeout or 0.0,
            )

        if scope.cancel_called:
            raise TimeoutError

        return response_msg

    finally:
        with suppress(Exception):
            await psub.unsubscribe()
            await psub.aclose()  # type: ignore[attr-defined]

publish_batch async #

publish_batch(cmd)
Source code in faststream/redis/publisher/producer.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@override
async def publish_batch(self, cmd: "RedisPublishCommand") -> int:
    batch = [
        cmd.message_format.encode(
            message=msg,
            correlation_id=cmd.correlation_id or "",
            reply_to=cmd.reply_to,
            headers=cmd.headers,
            serializer=self.serializer,
        )
        for msg in cmd.batch_bodies
    ]

    connection = cmd.pipeline or self._connection.client
    return await connection.rpush(cmd.destination, *batch)

connect #

connect(serializer=None)
Source code in faststream/redis/publisher/producer.py 
145
146
def connect(self, serializer: Optional["SerializerProto"] = None) -> None:
    self.serializer = serializer