Skip to content

# Module `liteyuki.comm.channel` ​

本模块定义了一个通用的通道类，用于进程间通信

### var `SYNC_ON_RECEIVE_FUNC` ​

- Description: 同步接收函数

- Type: `TypeAlias`

- Default: `Callable[[T], Any]`

### var `ASYNC_ON_RECEIVE_FUNC` ​

- Description: 异步接收函数

- Type: `TypeAlias`

- Default: `Callable[[T], Coroutine[Any, Any, Any]]`

### var `ON_RECEIVE_FUNC` ​

- Description: 接收函数

- Type: `TypeAlias`

- Default: `SYNC_ON_RECEIVE_FUNC | ASYNC_ON_RECEIVE_FUNC`

### var `SYNC_FILTER_FUNC` ​

- Description: 同步过滤函数

- Type: `TypeAlias`

- Default: `Callable[[T], bool]`

### var `ASYNC_FILTER_FUNC` ​

- Description: 异步过滤函数

- Type: `TypeAlias`

- Default: `Callable[[T], Coroutine[Any, Any, bool]]`

### var `FILTER_FUNC` ​

- Description: 过滤函数

- Type: `TypeAlias`

- Default: `SYNC_FILTER_FUNC | ASYNC_FILTER_FUNC`

### class `Channel(Generic[T])` ​

#### func `__init__(self, name: str, type_check: Optional[bool] = None)` ​

Description: 初始化通道

Arguments:

> - name: 通道ID
> - type_check: 是否开启类型检查, 若为空，则传入泛型默认开启，否则默认关闭

Source code or View on GitHub
python

    def __init__(self, name: str, type_check: Optional[bool]=None):
        self.conn_send, self.conn_recv = Pipe()
        self._conn_send_inner, self._conn_recv_inner = Pipe()
        self._closed = False
        self._on_main_receive_func_ids: list[int] = []
        self._on_sub_receive_func_ids: list[int] = []
        self.name: str = name
        self.is_receive_loop_running = False
        if type_check is None:
            type_check = self._get_generic_type() is not None
        elif type_check:
            if self._get_generic_type() is None:
                raise TypeError('Type hint is required for enforcing type check.')
        self.type_check = type_check
        if name in _channel:
            raise ValueError(f'Channel {name} already exists')
        if IS_MAIN_PROCESS:
            if name in _channel:
                raise ValueError(f'Channel {name} already exists')
            _channel[name] = self
            logger.debug(f'Channel {name} initialized in main process')
        else:
            logger.debug(f'Channel {name} initialized in sub process, should manually set in main process')

#### func `send(self, data: T)` ​

Description: 发送数据，发送函数为同步函数，没有异步的必要

Arguments:

> - data (T): 数据

Source code or View on GitHub
python

    def send(self, data: T):
        if self.type_check:
            _type = self._get_generic_type()
            if _type is not None and (not self._validate_structure(data, _type)):
                raise TypeError(f'Data must be an instance of {_type}, {type(data)} found')
        if self._closed:
            raise RuntimeError('Cannot send to a closed channel_')
        self.conn_send.send(data)

#### func `receive(self) -> T` ​

Description: 同步接收数据，会阻塞线程

Return: T: 数据

Source code or View on GitHub
python

    def receive(self) -> T:
        if self._closed:
            raise RuntimeError('Cannot receive from a closed channel_')
        while True:
            data = self.conn_recv.recv()
            return data

#### async func `async_receive(self) -> T` ​

Description: 异步接收数据，会挂起等待

Return: T: 数据

Source code or View on GitHub
python

    async def async_receive(self) -> T:
        loop = asyncio.get_running_loop()
        data = await loop.run_in_executor(None, self.receive)
        return data

#### func `on_receive(self, filter_func: Optional[FILTER_FUNC] = None) -> Callable[[Callable[[T], Any]], Callable[[T], Any]]` ​

Description: 接收数据并执行函数

Arguments:

> - filter_func (`Optional`[`FILTER_FUNC`], optional): 过滤函数. Defaults to None.

Return: Callable[[Callable[[T], Any]], Callable[[T], Any]]: 装饰器

Source code or View on GitHub
python

    def on_receive(self, filter_func: Optional[FILTER_FUNC]=None) -> Callable[[Callable[[T], Any]], Callable[[T], Any]]:
        if not IS_MAIN_PROCESS:
            raise RuntimeError('on_receive can only be used in main process')

        def decorator(func: Callable[[T], Any]) -> Callable[[T], Any]:
            global _func_id

            async def wrapper(data: T) -> Any:
                if filter_func is not None:
                    if is_coroutine_callable(filter_func):
                        if not await filter_func(data):
                            return
                    elif not filter_func(data):
                        return
                if is_coroutine_callable(func):
                    return await func(data)
                else:
                    return func(data)
            _callback_funcs[_func_id] = wrapper
            if IS_MAIN_PROCESS:
                self._on_main_receive_func_ids.append(_func_id)
            else:
                self._on_sub_receive_func_ids.append(_func_id)
            _func_id += 1
            return func
        return decorator

### var `active_channel` ​

- Description: 子进程可用的主动和被动通道

- Type: `Channel`

- Default: `Channel(name='active_channel')`

### var `passive_channel` ​

- Description: 被动通道

- Type: `Channel`

- Default: `Channel(name='passive_channel')`

### var `publish_channel` ​

- Description: 通道传递通道，主进程创建单例，子进程初始化时实例化

- Type: `Channel[tuple[str, dict[str, Any]]]`

- Default: `Channel(name='publish_channel')`

### var `channel_deliver_active_channel` ​

- Description: 通道传递通道，主进程创建单例，子进程初始化时实例化

- Type: `Channel[Channel[Any]]`

- Default: `NO_DEFAULT`

### var `channel_deliver_passive_channel` ​

- Description: 被动通道传递通道

- Type: `Channel[tuple[str, dict[str, Any]]]`

- Default: `NO_DEFAULT`

### func `set_channel(name: str, channel: Channel)` ​

Description: 设置通道实例

Arguments:

> - name (`str`): 通道名称
> - channel (`Channel`): 通道实例

Source code or View on GitHub
python

    def set_channel(name: str, channel: 'Channel'):
        if not isinstance(channel, Channel):
            raise TypeError(f'channel_ must be an instance of Channel, {type(channel)} found')
        if IS_MAIN_PROCESS:
            if name in _channel:
                raise ValueError(f'Channel {name} already exists')
            _channel[name] = channel
        else:
            channel_deliver_passive_channel.send(('set_channel', {'name': name, 'channel_': channel}))

### func `set_channels(channels: dict[str, Channel])` ​

Description: 设置通道实例

Arguments:

> - channels (`dict`[`str`, `Channel`]): 通道实例

Source code or View on GitHub
python

    def set_channels(channels: dict[str, 'Channel']):
        for name, channel in channels.items():
            set_channel(name, channel)

### func `get_channel(name: str) -> Channel` ​

Description: 获取通道实例

Arguments:

> - name (`str`): 通道名称

Return: `Channel`: 通道实例

Source code or View on GitHub
python

    def get_channel(name: str) -> 'Channel':
        if IS_MAIN_PROCESS:
            return _channel[name]
        else:
            recv_chan = Channel[Channel[Any]]('recv_chan')
            channel_deliver_passive_channel.send(('get_channel', {'name': name, 'recv_chan': recv_chan}))
            return recv_chan.receive()

### func `get_channels() -> dict[str, Channel]` ​

Description: 获取通道实例们

Return: `dict`[`str`, `Channel`]: 通道实例

Source code or View on GitHub
python

    def get_channels() -> dict[str, 'Channel']:
        if IS_MAIN_PROCESS:
            return _channel
        else:
            recv_chan = Channel[dict[str, Channel[Any]]]('recv_chan')
            channel_deliver_passive_channel.send(('get_channels', {'recv_chan': recv_chan}))
            return recv_chan.receive()

### var `channel_deliver_active_channel` ​

- Description: 主动通道传递通道

- Default: `Channel(name='channel_deliver_active_channel')`

### var `channel_deliver_passive_channel` ​

- Description: 被动通道传递通道

- Default: `Channel(name='channel_deliver_passive_channel')`

`@channel_deliver_passive_channel.on_receive(filter_func=lambda data: data[0] == 'set_channel')`

### func `on_set_channel(data: tuple[str, dict[str, Any]])` ​

Source code or View on GitHub
python

    @channel_deliver_passive_channel.on_receive(filter_func=lambda data: data[0] == 'set_channel')
    def on_set_channel(data: tuple[str, dict[str, Any]]):
        name, channel = (data[1]['name'], data[1]['channel_'])
        set_channel(name, channel)

`@channel_deliver_passive_channel.on_receive(filter_func=lambda data: data[0] == 'get_channel')`

### func `on_get_channel(data: tuple[str, dict[str, Any]])` ​

Source code or View on GitHub
python

    @channel_deliver_passive_channel.on_receive(filter_func=lambda data: data[0] == 'get_channel')
    def on_get_channel(data: tuple[str, dict[str, Any]]):
        name, recv_chan = (data[1]['name'], data[1]['recv_chan'])
        recv_chan.send(get_channel(name))

`@channel_deliver_passive_channel.on_receive(filter_func=lambda data: data[0] == 'get_channels')`

### func `on_get_channels(data: tuple[str, dict[str, Any]])` ​

Source code or View on GitHub
python

    @channel_deliver_passive_channel.on_receive(filter_func=lambda data: data[0] == 'get_channels')
    def on_get_channels(data: tuple[str, dict[str, Any]]):
        recv_chan = data[1]['recv_chan']
        recv_chan.send(get_channels())