Skip to content

FakeProducer

faststream.redis.test.FakeProducer #

FakeProducer(broker: RedisBroker)

Bases: RedisFastProducer

Source code in faststream/redis/test.py 
62
63
def __init__(self, broker: RedisBroker) -> None:
    self.broker = broker

broker instance-attribute #

broker = broker

publish async #

publish(
    message: SendableMessage,
    channel: Optional[str] = None,
    reply_to: str = "",
    headers: Optional[AnyDict] = None,
    correlation_id: Optional[str] = None,
    *,
    list: Optional[str] = None,
    stream: Optional[str] = None,
    rpc: bool = False,
    rpc_timeout: Optional[float] = 30.0,
    raise_timeout: bool = False
) -> Optional[Any]
Source code in faststream/redis/test.py 
 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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
async def publish(
    self,
    message: SendableMessage,
    channel: Optional[str] = None,
    reply_to: str = "",
    headers: Optional[AnyDict] = None,
    correlation_id: Optional[str] = None,
    *,
    list: Optional[str] = None,
    stream: Optional[str] = None,
    rpc: bool = False,
    rpc_timeout: Optional[float] = 30.0,
    raise_timeout: bool = False,
) -> Optional[Any]:
    any_of = channel or list or stream
    if any_of is None:
        raise ValueError(INCORRECT_SETUP_MSG)

    for handler in self.broker.handlers.values():  # pragma: no branch
        call = False
        batch = False

        if channel and (ch := handler.channel) is not None:
            call = bool(
                (not ch.pattern and ch.name == channel)
                or (
                    ch.pattern
                    and re.match(
                        ch.name.replace(".", "\\.").replace("*", ".*"),
                        channel,
                    )
                )
            )

        if list and (ls := handler.list_sub) is not None:
            batch = ls.batch
            call = list == ls.name

        if stream and (st := handler.stream_sub) is not None:
            batch = st.batch
            call = stream == st.name

        if call:
            r = await call_handler(
                handler=handler,
                message=build_message(
                    message=[message] if batch else message,
                    channel=any_of,
                    headers=headers,
                    correlation_id=correlation_id,
                    reply_to=reply_to,
                ),
                rpc=rpc,
                rpc_timeout=rpc_timeout,
                raise_timeout=raise_timeout,
            )

            if rpc:  # pragma: no branch
                return r

    return None

publish_batch async #

publish_batch(*msgs: SendableMessage, list: str) -> None
Source code in faststream/redis/test.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
async def publish_batch(
    self,
    *msgs: SendableMessage,
    list: str,
) -> None:
    for handler in self.broker.handlers.values():  # pragma: no branch
        if handler.list_sub and handler.list_sub.name == list:
            await call_handler(
                handler=handler,
                message=build_message(
                    message=msgs,
                    channel=list,
                ),
            )

    return None