Skip to content

FakeProducer

faststream.redis.testing.FakeProducer #

FakeProducer(broker: RedisBroker, config: ParserConfig)

Bases: RedisFastProducer

Source code in faststream/redis/testing.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
def __init__(self, broker: RedisBroker, config: ParserConfig) -> None:
    self.broker = broker

    default = RedisPubSubParser(config)

    self._parser = ParserComposition(
        broker._parser,
        default.parse_message,
    )
    self._decoder = ParserComposition(
        broker._decoder,
        default.decode_message,
    )

broker instance-attribute #

broker = broker

serializer instance-attribute #

serializer = serializer

publish async #

publish(cmd: RedisPublishCommand) -> int | bytes
Source code in faststream/redis/testing.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
@override
async def publish(self, cmd: "RedisPublishCommand") -> int | bytes:
    body = build_message(
        message=cmd.body,
        reply_to=cmd.reply_to,
        correlation_id=cmd.correlation_id or gen_cor_id(),
        headers=cmd.headers,
        message_format=cmd.message_format,
        serializer=self.broker.config.fd_config._serializer,
    )

    destination = _make_destination_kwargs(cmd)
    visitors = (ChannelVisitor(), ListVisitor(), StreamVisitor())

    for handler in self.broker.subscribers:  # pragma: no branch
        handler = cast("LogicSubscriber", handler)
        for visitor in visitors:
            if visited_ch := visitor.visit(**destination, sub=handler):
                msg = visitor.get_message(
                    visited_ch,
                    body,
                    handler,  # type: ignore[arg-type]
                )

                await self._execute_handler(msg, handler)

    return 0

request async #

request(cmd: RedisPublishCommand) -> PubSubMessage
Source code in faststream/redis/testing.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
@override
async def request(self, cmd: "RedisPublishCommand") -> "PubSubMessage":
    body = build_message(
        message=cmd.body,
        correlation_id=cmd.correlation_id or gen_cor_id(),
        headers=cmd.headers,
        message_format=cmd.message_format,
        serializer=self.broker.config.fd_config._serializer,
    )

    destination = _make_destination_kwargs(cmd)
    visitors = (ChannelVisitor(), ListVisitor(), StreamVisitor())

    for handler in self.broker.subscribers:  # pragma: no branch
        handler = cast("LogicSubscriber", handler)
        for visitor in visitors:
            if visited_ch := visitor.visit(**destination, sub=handler):
                msg = visitor.get_message(
                    visited_ch,
                    body,
                    handler,  # type: ignore[arg-type]
                )

                with anyio.fail_after(cmd.timeout):
                    return await self._execute_handler(msg, handler)

    raise SubscriberNotFound

publish_batch async #

publish_batch(cmd: RedisPublishCommand) -> int
Source code in faststream/redis/testing.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
@override
async def publish_batch(self, cmd: "RedisPublishCommand") -> int:
    data_to_send = [
        build_message(
            m,
            correlation_id=cmd.correlation_id or gen_cor_id(),
            headers=cmd.headers,
            message_format=cmd.message_format,
            serializer=self.broker.config.fd_config._serializer,
        )
        for m in cmd.batch_bodies
    ]

    visitor = ListVisitor()
    for handler in self.broker.subscribers:  # pragma: no branch
        handler = cast("LogicSubscriber", handler)
        if visitor.visit(list=cmd.destination, sub=handler):
            casted_handler = cast("_ListHandlerMixin", handler)

            if casted_handler.list_sub.batch:
                msg = visitor.get_message(
                    channel=cmd.destination,
                    body=data_to_send,
                    sub=casted_handler,
                )

                await self._execute_handler(msg, handler)

    return 0

connect #

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