Onionr/src/onionrthreads/__init__.py

57 lines
1.5 KiB
Python
Raw Normal View History

from typing import Callable
from typing import Iterable
import traceback
from threading import Thread
2020-11-15 18:26:25 +00:00
from uuid import uuid4
2020-07-31 01:15:36 +00:00
from time import sleep
from logger import log as logging
2022-02-10 01:28:31 +00:00
def _onionr_thread(func: Callable,
sleep_secs: int, initial_sleep, *args, **kwargs):
2020-11-15 18:26:25 +00:00
thread_id = str(uuid4())
2020-07-31 01:15:36 +00:00
if initial_sleep:
sleep(initial_sleep)
while True:
try:
2022-02-10 01:28:31 +00:00
func(*args, **kwargs)
except Exception as _: # noqa
logging.warn(
2020-11-15 18:26:25 +00:00
f"Onionr thread exception in {thread_id} \n" +
traceback.format_exc(),
)
2020-07-31 01:15:36 +00:00
sleep(sleep_secs)
def add_onionr_thread(
2022-02-10 01:28:31 +00:00
func: Callable,
sleep_secs: int, thread_name: str,
*args, initial_sleep: int = 5, **kwargs):
"""Spawn a new onionr thread that exits when the main thread does.
Runs in an infinite loop with sleep between calls
Passes in an interable args and sleep variables"""
2022-02-10 01:28:31 +00:00
Thread(target=_onionr_thread,
2022-02-10 01:28:31 +00:00
args=(
func,
sleep_secs,
initial_sleep,
*args),
kwargs=kwargs,
name=thread_name,
2022-02-10 01:28:31 +00:00
daemon=True).start()
2022-03-13 01:28:18 +00:00
def add_delayed_thread(func: Callable, sleep_secs: int, *args, **kwargs):
assert sleep_secs > 0
def _do_delay_thread():
t = Thread(target=func, args=args, kwargs=kwargs, daemon=True)
sleep(sleep_secs)
t.start()
t = Thread(target=_do_delay_thread, daemon=True)
2022-03-13 01:28:18 +00:00
t.start()
return t