| 21 | |
| 22 | # todo 线程安全 |
| 23 | class ThreadBasedSession(Session): |
| 24 | thread2session = {} # thread_id -> session |
| 25 | |
| 26 | unhandled_task_mq_maxsize = 1000 |
| 27 | event_mq_maxsize = 100 |
| 28 | callback_mq_maxsize = 100 |
| 29 | |
| 30 | @classmethod |
| 31 | def get_current_session(cls) -> "ThreadBasedSession": |
| 32 | curr = id(threading.current_thread()) |
| 33 | session = cls.thread2session.get(curr) |
| 34 | if session is None: |
| 35 | raise SessionNotFoundException("Can't find current session. " |
| 36 | "Maybe session closed or forget to use `register_thread()`.") |
| 37 | return session |
| 38 | |
| 39 | @classmethod |
| 40 | def get_current_task_id(cls): |
| 41 | return cls._get_task_id(threading.current_thread()) |
| 42 | |
| 43 | @staticmethod |
| 44 | def _get_task_id(thread: threading.Thread): |
| 45 | tname = getattr(thread, '_target', 'task') |
| 46 | tname = getattr(tname, '__name__', tname) |
| 47 | return '%s-%s' % (tname, id(thread)) |
| 48 | |
| 49 | def __init__(self, target, session_info, on_task_command=None, on_session_close=None, loop=None): |
| 50 | """ |
| 51 | :param target: 会话运行的函数. 为None时表示Script mode |
| 52 | :param on_task_command: 当Task内发送Command给session的时候触发的处理函数 |
| 53 | :param on_session_close: 会话结束的处理函数 |
| 54 | :param loop: 事件循环。若 on_task_command 或者 on_session_close 中有调用使用asyncio事件循环的调用， |
| 55 | 则需要事件循环实例来将回调在事件循环的线程中执行 |
| 56 | """ |
| 57 | assert target is None or (not iscoroutinefunction(target)) and (not isgeneratorfunction(target)), ValueError( |
| 58 | "ThreadBasedSession only accept a simple function as task function, " |
| 59 | "not coroutine function or generator function. ") |
| 60 | |
| 61 | super().__init__(session_info) |
| 62 | |
| 63 | self._on_task_command = on_task_command or (lambda _: None) |
| 64 | self._on_session_close = on_session_close or (lambda: None) |
| 65 | self._loop = loop |
| 66 | |
| 67 | self.threads = [] # 注册到当前会话的线程集合 |
| 68 | self.unhandled_task_msgs = LimitedSizeQueue(maxsize=self.unhandled_task_mq_maxsize) |
| 69 | |
| 70 | self.task_mqs = {} # task_id -> event msg queue |
| 71 | self._closed = False |
| 72 | |
| 73 | # 用于实现回调函数的注册 |
| 74 | self.callback_mq = None |
| 75 | self.callback_thread = None |
| 76 | self.callbacks = {} # callback_id -> (callback_func, is_mutex) |
| 77 | |
| 78 | if target is not None: |
| 79 | self._start_main_task(target) |
| 80 |
no outgoing calls
no test coverage detected
searching dependent graphs…