56 lines
2 KiB
Python
56 lines
2 KiB
Python
#!/usr/bin/env python3
|
|
|
|
# Libervia AsyncIO helper methods
|
|
# Copyright (C) 2009-2023 Jérôme Poisson (goffi@goffi.org)
|
|
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
import asyncio
|
|
from typing import Any, Coroutine
|
|
|
|
from libervia.backend.core import log as logging
|
|
|
|
log = logging.getLogger(__name__)
|
|
background_tasks = set()
|
|
|
|
|
|
def _on_task_done(task: asyncio.Future) -> None:
|
|
"""Callback function to execute when a task is done.
|
|
|
|
@param task: The completed task.
|
|
Note: The function removes the task from the tracking set and logs any exceptions
|
|
that might have occurred.
|
|
"""
|
|
background_tasks.discard(task)
|
|
e = task.exception()
|
|
if e is not None:
|
|
exc_info = (type(e), e, e.__traceback__)
|
|
log.error("Task failed:", exc_info=exc_info)
|
|
|
|
|
|
def run_async(async_method: Coroutine | asyncio.Future) -> None:
|
|
"""Schedules and tracks an asynchronous method.
|
|
|
|
@param async_method: The method to be scheduled for execution.
|
|
Note: The function keeps a strong reference to the task to prevent it from being
|
|
garbage-collected. In case of task failure, logs the error.
|
|
cf. https://docs.python.org/3/library/asyncio-task.html#asyncio.create_task
|
|
"""
|
|
if isinstance(async_method, asyncio.Future):
|
|
task = asyncio.ensure_future(async_method)
|
|
else:
|
|
task = asyncio.create_task(async_method)
|
|
|
|
background_tasks.add(task)
|
|
task.add_done_callback(_on_task_done)
|