snowykami/testpage
1
0
Fork 0
forked from bot/app
Code Pull Requests Activity

✨ 添加对主流框架的消息io支持

Browse Source
This commit is contained in:
远野千束
2024-08-20 06:20:41 +08:00
parent 237789e0d4
commit 0c942d9806
26 changed files with 267 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
liteyuki/comm/channel.py
View File

@ -60,7 +60,7 @@ class Channel(Generic[T]):
elif type_check:
if self._get_generic_type() is None:
raise TypeError("Type hint is required for enforcing type_ check.")
raise TypeError("Type hint is required for enforcing type check.")
self.type_check = type_check
def _get_generic_type(self) -> Optional[type]:
@ -158,7 +158,7 @@ class Channel(Generic[T]):
async def wrapper(data: T) -> Any:
if filter_func is not None:
if is_coroutine_callable(filter_func):
if not (await filter_func(data)): # type_: ignore
if not (await filter_func(data)): # type: ignore
return
else:
if not filter_func(data):

10
liteyuki/comm/storage.py
View File

@ -13,9 +13,9 @@ from liteyuki.utils import IS_MAIN_PROCESS, is_coroutine_callable, run_coroutine
if IS_MAIN_PROCESS:
_locks = {}
_on_main_subscriber_receive_funcs: dict[str, list[ASYNC_ON_RECEIVE_FUNC]] = {} # type_: ignore
_on_main_subscriber_receive_funcs: dict[str, list[ASYNC_ON_RECEIVE_FUNC]] = {} # type: ignore
"""主进程订阅者接收函数"""
_on_sub_subscriber_receive_funcs: dict[str, list[ASYNC_ON_RECEIVE_FUNC]] = {} # type_: ignore
_on_sub_subscriber_receive_funcs: dict[str, list[ASYNC_ON_RECEIVE_FUNC]] = {} # type: ignore
"""子进程订阅者接收函数"""
@ -169,7 +169,7 @@ class KeyValueStore:
(
"publish",
{
"channel_": channel_,
"channel": channel_,
"data" : data
}
)
@ -234,7 +234,7 @@ class KeyValueStore:
data = self.active_chan.receive()
if data[0] == "publish":
# 运行主进程订阅函数
self.run_subscriber_receive_funcs(data[1]["channel_"], data[1]["data"])
self.run_subscriber_receive_funcs(data[1]["channel"], data[1]["data"])
# 推送给子进程
self.publish_channel.send(data)
else:
@ -242,7 +242,7 @@ class KeyValueStore:
data = self.publish_channel.receive()
if data[0] == "publish":
# 运行子进程订阅函数
self.run_subscriber_receive_funcs(data[1]["channel_"], data[1]["data"])
self.run_subscriber_receive_funcs(data[1]["channel"], data[1]["data"])
class GlobalKeyValueStore: