注册一个任务供生命周期管理
(self, task: asyncio.Task, description: str = None)
| 33 | log.debug("TaskManager initialized") |
| 34 | |
| 35 | def register_task(self, task: asyncio.Task, description: str = None) -> asyncio.Task: |
| 36 | """注册一个任务供生命周期管理""" |
| 37 | self._tasks.add(task) |
| 38 | task.add_done_callback(lambda t: self._tasks.discard(t)) |
| 39 | |
| 40 | if description: |
| 41 | task.set_name(description) |
| 42 | |
| 43 | log.debug(f"Registered task: {task.get_name() or 'unnamed'}") |
| 44 | return task |
| 45 | |
| 46 | def create_task(self, coro, *, name: str = None) -> asyncio.Task: |
| 47 | """创建并注册一个任务""" |