Source code for ska_tango_base.executor._future

from __future__ import annotations

import typing

import ska_control_model as scm

from .. import type_hints
from .executor import DEFAULT_MAX_QUEUED_TASKS, TaskExecutor




class TaskExecutorComponentManager:
    """A component manager with support for asynchronous tasking."""



    def __init__(
        self,
        *args: typing.Any,
        max_queue_size: int = DEFAULT_MAX_QUEUED_TASKS,
        **kwargs: typing.Any,
    ) -> None:
        """
        Initialise a new TaskExecutorComponentManager instance.

        :param args: Additional positional arguments.
        :param max_queue_size: Optional maximum size of the tasks input queue
            (minimum 1).
        :param kwargs: Additional keyword arguments.
        """
        self._task_executor = TaskExecutor(
            unhandled_exception_callback=self._on_unhandled_exception,
            max_queued_tasks=max_queue_size,
        )
        super().__init__(*args, **kwargs)




    def cleanup(self) -> None:
        """Shutdown the task executor."""
        self._task_executor.shutdown()


    def _on_unhandled_exception(self, exception: Exception) -> None:
        """
        Do something when a task raises an unhandled exception.

        :param exception: the unhandled exception that was caught.
        """

    @property
    def max_executing_tasks(self) -> int:
        """
        Get the maximum number of simultaneously executing tasks.

        :return: The maximum number of simultaneously executing tasks.
        """
        return self._task_executor._max_executing_tasks

    @property
    def max_queued_tasks(self) -> int:
        """
        Get the task queue size.

        :return: The task queue size
        """
        return self._task_executor.max_queued_tasks



    def submit_task(
        self,
        func: typing.Any,
        args: typing.Any = None,
        kwargs: typing.Any = None,
        is_cmd_allowed: typing.Callable[[], bool] | None = None,
        task_callback: type_hints.TaskCallbackType | None = None,
    ) -> tuple[scm.TaskStatus, str]:
        """
        Submit a task to the task executor.

        :param func: function/bound method to be run
        :param args: positional arguments to the function
        :param kwargs: keyword arguments to the function
        :param is_cmd_allowed: sanity check for func
        :param task_callback: callback to be called whenever the status
            of the task changes.

        :return: tuple of TaskStatus & message
        """
        return self._task_executor.submit(
            func, args, kwargs, is_cmd_allowed, task_callback=task_callback
        )




    def abort_tasks(
        self,
        task_callback: type_hints.TaskCallbackType | None = None,
    ) -> tuple[scm.TaskStatus, str]:
        """
        Tell the task executor to abort all tasks.

        :param task_callback: callback to be called whenever the status
            of this abort task changes.

        :return: tuple of TaskStatus & message
        """
        return self._task_executor.abort(task_callback)