Skip to content

ChannelManagerImpl

faststream.rabbit.helpers.channel_manager.ChannelManagerImpl #

ChannelManagerImpl(
    default_channel: Optional[Channel] = None,
)

Bases: ChannelManager

Source code in faststream/rabbit/helpers/channel_manager.py 
45
46
47
48
49
50
51
52
53
def __init__(
    self,
    default_channel: Optional["Channel"] = None,
) -> None:
    self.__connection: ConnectionState = EmptyConnectionState()

    self.__default_channel = default_channel or Channel()

    self.__channels: dict[Channel, aio_pika.RobustChannel] = {}

connect #

connect(connection: RobustConnection) -> None
Source code in faststream/rabbit/helpers/channel_manager.py 
55
56
def connect(self, connection: "aio_pika.RobustConnection") -> None:
    self.__connection = ConnectedState(connection)

disconnect #

disconnect() -> None
Source code in faststream/rabbit/helpers/channel_manager.py 
58
59
60
def disconnect(self) -> None:
    self.__connection = EmptyConnectionState()
    self.__channels.clear()

get_channel async #

get_channel(
    channel: Optional[Channel] = None,
) -> RobustChannel
Source code in faststream/rabbit/helpers/channel_manager.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
async def get_channel(
    self,
    channel: Optional["Channel"] = None,
) -> "aio_pika.RobustChannel":
    if channel is None:
        channel = self.__default_channel

    if (ch := self.__channels.get(channel)) is None:
        self.__channels[channel] = ch = cast(
            "aio_pika.RobustChannel",
            await self.__connection.connection.channel(
                channel_number=channel.channel_number,
                publisher_confirms=channel.publisher_confirms,
                on_return_raises=channel.on_return_raises,
            ),
        )

        if channel.prefetch_count:
            await ch.set_qos(
                prefetch_count=channel.prefetch_count,
                global_=channel.global_qos,
            )

    return ch