bot/croterline
1
0
Fork 0
mirror of https://github.com/LiteyukiStudio/croterline.git synced 2024-11-25 01:55:05 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

✨ first commit

Browse Source
This commit is contained in:
远野千束 2024-10-13 17:28:37 +08:00
parent 5cb3424ad0
commit e151a7af0c
2 changed files with 18 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
croterline/process.py
View File

@ -1,10 +1,12 @@
from multiprocessing import Process as _Process
from typing import Callable, Any
from mypy.nodes import TypeAlias
from croterline.context import Context
from croterline.utils import IsMainProcess
type ProcessFuncType = Callable[[tuple[Any, ...], dict[str, Any]], None]
ProcessFuncType: TypeAlias = Callable[[tuple[Any, ...], dict[str, Any]], None]
_current_ctx: "Context | None" = None # 注入当前进程上下文
@ -49,6 +51,8 @@ def get_ctx() -> Context | None:
def _wrapper(func: ProcessFuncType, ctx: Context, *args, **kwargs):
print("args", args)
print("kwargs", kwargs)
global _current_ctx
_current_ctx = ctx

16
tests/test_subprocess.py
View File

@ -5,6 +5,8 @@ from croterline.process import SubProcess, get_ctx
def p_func(*args, **kwargs):
print("Args", args)
print("Kwargs", kwargs)
i = 0
ctx = get_ctx()
while True:
@ -15,10 +17,16 @@ def p_func(*args, **kwargs):
ctx.sub_chan << "end"
def p_func2(*args, **kwargs):
print("args: ", args)
print("kwargs: ", kwargs)
raise Exception("Test")
class TestSubProcess:
def test_run(self):
print("start")
sp = SubProcess("test", p_func, Context())
sp = SubProcess("test", p_func, Context(), 1, 2, 3, k1=1, k2=2)
sp.start()
while True:
@ -30,5 +38,7 @@ class TestSubProcess:
print("finished")
sp.terminate()
def test_decorator(self):
pass
def test_input(self):
print("test_input")
sp = SubProcess("test2", p_func2, Context(), 1, 2, 3, host=1, port=2)
sp.start()