get_broker_channels

def get_broker_channels(
    broker: "BrokerUsecase[MsgType, ConnectionType, Any]",
    servers: list[str] | None = None,
) -> tuple[dict[str, Channel], dict[str, Operation]]:
    """Get the broker channels for an application."""
    channels = {}
    operations = {}

    channel_servers = [
        ref("servers", server_name) for server_name in (servers or ())
    ] or None

    for sub in filter(lambda s: s.specification.include_in_schema, broker.subscribers):
        sub_schema = sub.schema()

        # A title cannot name the operations of several channels: written once
        # per channel it leaves one operation and the rest unreferenced.
        sub_title = sub.specification.config.title_ if len(sub_schema) == 1 else None

        for sub_key, sub_channel in sub_schema.items():
            channel_obj = Channel.from_sub(sub_channel, servers=channel_servers)

            channel_key = clear_key(sub_key)
            if channel_key in channels:
                warnings.warn(
                    f"Overwrite channel handler, channels have the same names: `{channel_key}`",
                    RuntimeWarning,
                    stacklevel=1,
                )

            channels[channel_key] = channel_obj

            operation_key = (
                f"{channel_key}Subscribe"
                if sub_title is None or sub_title == "/"
                else sub_title
            )
            if operation_key in operations:
                warnings.warn(
                    f"Overwrite channel handler, operations have the same names: `{operation_key}`",
                    RuntimeWarning,
                    stacklevel=1,
                )

            operations[operation_key] = Operation.from_sub(
                messages=[
                    Reference(**ref("channels", channel_key, "messages", msg_name))
                    for msg_name in channel_obj.messages
                ],
                channel=Reference(**ref("channels", channel_key)),
                operation=sub_channel.operation,
            )

    for pub in filter(lambda p: p.specification.include_in_schema, broker.publishers):
        for pub_key, pub_channel in pub.schema().items():
            channel_obj = Channel.from_pub(pub_channel, servers=channel_servers)

            channel_key = clear_key(pub_key)
            if channel_key in channels:
                warnings.warn(
                    f"Overwrite channel handler, channels have the same names: `{channel_key}`",
                    RuntimeWarning,
                    stacklevel=1,
                )
            channels[channel_key] = channel_obj

            operations[channel_key] = Operation.from_pub(
                messages=[
                    Reference(**ref("channels", channel_key, "messages", msg_name))
                    for msg_name in channel_obj.messages
                ],
                channel=Reference(**ref("channels", channel_key)),
                operation=pub_channel.operation,
            )

    return channels, operations