Skip to content

FakeProducer

faststream.nats.test.FakeProducer #

FakeProducer(broker: NatsBroker)

Bases: NatsFastProducer

Source code in faststream/nats/test.py 
49
50
def __init__(self, broker: NatsBroker) -> None:
    self.broker = broker

broker instance-attribute #

broker = broker

publish async #

publish(message: SendableMessage, subject: str, reply_to: str = '', headers: Optional[Dict[str, str]] = None, correlation_id: Optional[str] = None, timeout: Optional[float] = None, stream: Optional[str] = None, *, rpc: bool = False, rpc_timeout: Optional[float] = None, raise_timeout: bool = False) -> Optional[SendableMessage]
Source code in faststream/nats/test.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 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
102
103
104
105
106
107
108
109
110
111
112
113
114
@override
async def publish(  # type: ignore[override]
    self,
    message: SendableMessage,
    subject: str,
    reply_to: str = "",
    headers: Optional[Dict[str, str]] = None,
    correlation_id: Optional[str] = None,
    # NatsJSFastProducer compatibility
    timeout: Optional[float] = None,
    stream: Optional[str] = None,
    *,
    rpc: bool = False,
    rpc_timeout: Optional[float] = None,
    raise_timeout: bool = False,
) -> Optional[SendableMessage]:
    incoming = build_message(
        message=message,
        subject=subject,
        headers=headers,
        correlation_id=correlation_id,
        reply_to=reply_to,
    )

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

        if stream and getattr(handler.stream, "name", None) != stream:
            continue

        if subject == handler.subject:
            call = True

        else:
            call = True

            for current, base in zip_longest(
                subject.split("."),
                handler.subject.split("."),
                fillvalue=None,
            ):
                if base == ">":
                    break

                if base != "*" and current != base:
                    call = False
                    break

        if call:
            r = await call_handler(
                handler=handler,
                message=[incoming]
                if getattr(handler.pull_sub, "batch", False)
                else incoming,
                rpc=rpc,
                rpc_timeout=rpc_timeout,
                raise_timeout=raise_timeout,
            )

            if rpc:
                return r

    return None