Skip to content

process_msg

faststream.broker.utils.process_msg async #

process_msg(msg: Literal[None], middlewares: Iterable[BrokerMiddleware[MsgType]], parser: Callable[[MsgType], Awaitable[StreamMessage[MsgType]]], decoder: Callable[[StreamMessage[MsgType]], Any]) -> None

process_msg(msg: MsgType, middlewares: Iterable[BrokerMiddleware[MsgType]], parser: Callable[[MsgType], Awaitable[StreamMessage[MsgType]]], decoder: Callable[[StreamMessage[MsgType]], Any]) -> StreamMessage[MsgType]

process_msg(msg, middlewares, parser, decoder)
Source code in faststream/broker/utils.py 
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
async def process_msg(
    msg: Optional[MsgType],
    middlewares: Iterable["BrokerMiddleware[MsgType]"],
    parser: Callable[[MsgType], Awaitable["StreamMessage[MsgType]"]],
    decoder: Callable[["StreamMessage[MsgType]"], "Any"],
) -> Optional["StreamMessage[MsgType]"]:
    if msg is None:
        return None

    async with AsyncExitStack() as stack:
        return_msg: Callable[
            [StreamMessage[MsgType]],
            Awaitable[StreamMessage[MsgType]],
        ] = return_input

        for m in middlewares:
            mid = m(msg)
            await stack.enter_async_context(mid)
            return_msg = partial(mid.consume_scope, return_msg)

        parsed_msg = await parser(msg)
        parsed_msg._decoded_body = await decoder(parsed_msg)
        return await return_msg(parsed_msg)

    raise AssertionError("unreachable")