kev/Onionr
1
0
Fork 0
Code Issues 24 Pull Requests Projects Releases 2 Wiki Activity

Adjusted onionrthreads to take kwargs

Browse Source
This commit is contained in:
Kevin F 2022-02-09 19:28:31 -06:00
parent 4fba79950c
commit e55beec18c
2 changed files with 15 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
src/onionrthreads/__init__.py
View File

@ -10,14 +10,14 @@ from time import sleep
import logger import logger
def _onionr_thread(func: Callable, args: Iterable, def _onionr_thread(func: Callable,
sleep_secs: int, initial_sleep): sleep_secs: int, initial_sleep, *args, **kwargs):
thread_id = str(uuid4()) thread_id = str(uuid4())
if initial_sleep: if initial_sleep:
sleep(initial_sleep) sleep(initial_sleep)
while True: while True:
try: try:
func(*args) func(*args, **kwargs)
except Exception as _: # noqa except Exception as _: # noqa
logger.warn( logger.warn(
f"Onionr thread exception in {thread_id} \n" + f"Onionr thread exception in {thread_id} \n" +
@ -27,11 +27,18 @@ def _onionr_thread(func: Callable, args: Iterable,
def add_onionr_thread( def add_onionr_thread(
func: Callable, args: Iterable, func: Callable,
sleep_secs: int, initial_sleep: int = 5): sleep_secs: int, *args, initial_sleep: int = 5, **kwargs):
"""Spawn a new onionr thread that exits when the main thread does. """Spawn a new onionr thread that exits when the main thread does.
Runs in an infinite loop with sleep between calls Runs in an infinite loop with sleep between calls
Passes in an interable args and sleep variables""" Passes in an interable args and sleep variables"""
Thread(target=_onionr_thread, Thread(target=_onionr_thread,
args=(func, args, sleep_secs, initial_sleep), daemon=True).start() args=(
func,
sleep_secs,
initial_sleep,
*args),
kwargs=kwargs,
daemon=True).start()

5
tests/test_onionr_threads.py
View File

@ -25,7 +25,7 @@ class OnionrThreadsTests(unittest.TestCase):
def _test_func(obj_list): def _test_func(obj_list):
obj_list.append(1) obj_list.append(1)
add_onionr_thread(_test_func, (l,), 0.05, 0) add_onionr_thread(_test_func, 0.05, l, initial_sleep=0)
sleep(0.05) sleep(0.05)
self.assertGreaterEqual(len(l), 1) self.assertGreaterEqual(len(l), 1)
@ -34,8 +34,7 @@ class OnionrThreadsTests(unittest.TestCase):
def _test_func(obj_list): def _test_func(obj_list):
obj_list.append(1) obj_list.append(1)
add_onionr_thread(_test_func, (l,), 0.05, 0.1) add_onionr_thread(_test_func, 0.05, l, initial_sleep=1)
sleep(0.06)
self.assertEqual(len(l), 0) self.assertEqual(len(l), 0)