8.14. eventloop 模块 API 文档

8.14.1. API

8.14.1.1. class EventTask:

Task Item of EventLoop

def __init__(self,func,callback,args,period_ms,delay_ms):...

:param func: function to be called

:param callback: callback function

:param args: arguments of func

:param period_ms: period of periodic task

8.14.1.2. class EventLoop:

Event Loop

def __init__(self,period_ms=100,thread_stack=0):...

:param period_ms: period of loop

:param thread_stack: stack size of thread

def _add_task(self,task_name,func,callback,args,period_ms,delay_ms):...

def start_new_task(self,func,args,is_periodic=True,period_ms=1000,callback=None,task_name=None,delay_ms=None):...

Add a task to EventLoop

:param task_name: name of task

:param func: function to be called

:param period_ms: period of task

:param callback: callback function

:param args: arguments of func

def start_new_task_once(self,func,args,callback=None,task_name=None,delay_ms=None):...

Add a task to EventLoop, run once

:param task_name: name of task

:param func: function to be called

:param callback: callback function

:param args: arguments of func

def start_new_task_periodic(self,func,args,period_ms=1000,callback=None,task_name=None,delay_ms=None):...

Add a task to EventLoop, run periodically

:param task_name: name of task

:param func: function to be called

:param period_ms: period of task

:param callback: callback function

:param args: arguments of func

def remove_task(self,task_name):...

Remove a task from EventLoop

:param task_name: name of task

def _run_task(self,task:EventTask):...

def _run_thread(self):...

def start(self):...

Start EventLoop

def stop(self):...

Stop EventLoop

def set_debug(enable:bool):...

def _debug(*args):...

def _get_default_event_loop():...

def start_new_task(func,args,is_periodic=True,period_ms=1000,callback=None,task_name=None,delay_ms=None):...

Add a task to EventLoop

:param task_name: name of task

:param func: function to be called

:param period_ms: period of task

:param callback: callback function

:param args: arguments of func

def start_new_task_once(func,args,callback=None,task_name=None,delay_ms=None):...

Add a task to EventLoop, run once

:param task_name: name of task

:param func: function to be called

:param callback: callback function

:param args: arguments of func

def start_new_task_periodic(func,args,period_ms=1000,callback=None,task_name=None,delay_ms=None):...

Add a task to EventLoop, run periodically

:param task_name: name of task

:param func: function to be called

:param period_ms: period of task

:param callback: callback function

:param args: arguments of func

def remove_task(task_name):...

Remove a task from EventLoop

:param task_name: name of task

def stop():...

Stop EventLoop

def start():...

Start EventLoop

def __del__():...

8.14.2. Examples

8.14.2.1. test2.py

import time
from eventloop import EventLoop

run_time = 0

eventloop.set_debug(True)


def test_func(arg1, arg2):
    global run_time
    run_time += 1
    print("Running test function with arguments:", arg1, arg2)
    return arg1 + arg2


def test_func2(arg1, arg2):
    print("test function 2 with arguments:", arg1, arg2)
    return arg1 + arg2


def test_callback(res):
    print("Running test callback function", res)
    assert res == "Hello World"


# Test case 2: Add and run a periodic task
event_loop = EventLoop(period_ms=10)

event_loop.start_new_task_periodic(
    test_func, ("Hello", " World"),
    period_ms=100,
    callback=test_callback
)

event_loop.start_new_task(
    test_func2, ("Hello", " World"),
    is_periodic=True,
    period_ms=20,
    callback=test_callback
)

event_loop.start()

# Sleep for enough time to allow the periodic task to run multiple times
while run_time < 3:
    time.sleep(0.1)

event_loop.stop()

# Test case 3: Test removing a task
event_loop.start_new_task_periodic(
    test_func, ("Hello", " World"), callback=test_callback, task_name="test_task_remove")
event_loop.remove_task("test_task_remove")

print(event_loop._tasks)
assert "test_task_remove" not in event_loop._tasks, "Failed to remove the task"

8.14.2.2. once2.py

import eventloop
from PikaStdLib import MemChecker
import time

eventloop._is_debug = True
expect_finished = 10
finished = 0

def test_func(arg1, arg2):
    global finished
    finished += 1
    print("finished:", finished)
    print("Running test function with arguments:", arg1, arg2)
    MemChecker().now()


MemChecker().now()
for i in range(expect_finished):
    eventloop.start_new_task_once(
        test_func, ("Hello", " World"), delay_ms=10)
MemChecker().now()

while finished < expect_finished:
    time.sleep(0.1)
eventloop.stop()

8.14.2.3. once1.py

import eventloop
from PikaStdLib import MemChecker
import time

eventloop._is_debug = True
expect_finished = 10
finished = 0

def test_func(arg1, arg2):
    global finished
    finished += 1
    print("finished:", finished)
    print("Running test function with arguments:", arg1, arg2)
    MemChecker().now()


MemChecker().now()
for i in range(expect_finished):
    eventloop.start_new_task_once(
        test_func, ("Hello", " World"))
MemChecker().now()

while finished < expect_finished:
    time.sleep(0.1)
eventloop.stop()

8.14.2.4. delay1.py

import time
from eventloop import EventLoop

run_time = 0

eventloop.set_debug(True)


def test_func(arg1, arg2):
    global run_time
    run_time += 1
    print("Running test function with arguments:", arg1, arg2)
    return arg1 + arg2


def test_func2(arg1, arg2):
    print("test function 2 with arguments:", arg1, arg2)
    return arg1 + arg2


def test_callback(res):
    print("Running test callback function", res)
    assert res == "Hello World"


# Test case 2: Add and run a periodic task

eventloop.start_new_task_once(
    test_func, ("Hello", " World"),
    callback=test_callback,
    delay_ms=200
)

eventloop.start_new_task(
    test_func2, ("Hello", " World"),
    is_periodic=True,
    period_ms=20,
    callback=test_callback
)

# Sleep for enough time to allow the periodic task to run multiple times
while run_time < 1:
    time.sleep(0.1)

8.14.2.5. test1.py

import time
import eventloop
from eventloop import EventLoop

finished = False

eventloop.set_debug(True)


def test_func(arg1, arg2):
    print("Running test function with arguments:", arg1, arg2)
    return arg1 + arg2


def test_callback(res):
    global finished
    print("Running test callback function", res)
    assert res == "Hello World"
    finished = True


# Test case 1: Add and run a one-time task
event_loop = EventLoop(period_ms=10)

event_loop.start_new_task_once(
    test_func, ("Hello", " World"), callback=test_callback)
event_loop.start()

# Sleep for enough time to allow the one-time task to run
while not finished:
    time.sleep(0.1)

event_loop.stop()

8.14.2.6. test3.py

import time
import eventloop

run_time = 0

eventloop.set_debug(True)


def test_func(arg1, arg2):
    global run_time
    run_time += 1
    print("Running test function with arguments:", arg1, arg2)
    return arg1 + arg2


def test_func2():
    print("test function 2")


def test_func3(arg):
    print("test function 3 with argument:", arg)


def test_callback(res):
    print("Running test callback function", res)
    assert res == "Hello World"


eventloop.start_new_task(test_func, ("Hello", " World"))

eventloop.start_new_task_periodic(
    test_func2, (),
    period_ms=20
)

eventloop.start_new_task_once(test_func3, ("Hello"))

# Sleep for enough time to allow the periodic task to run multiple times
while run_time < 3:
    time.sleep(0.1)

# Test case 3: Test removing a task
eventloop.start_new_task_periodic(
    test_func, ("Hello", " World"), callback=test_callback, task_name="test_task_remove")
eventloop.remove_task("test_task_remove")