Source code for edri.abstract.worker.worker_thread

from abc import ABC
from multiprocessing.connection import Connection
from threading import Thread
from typing import Any, TypeVar

from edri.dataclass.event import Event
from edri.abstract.worker import Worker

__all__ = ["WorkerThread", "T"]


T = TypeVar('T', bound=Event)




class WorkerThread(Worker[T], Thread, ABC):
    """
    A WorkerThread class that combines the custom Worker class with threading
    capabilities, designed to run as a separate thread. This class is abstract
    and should be subclassed to implement specific worker behavior in a threaded
    environment.

    Note:
        As this class is abstract, it cannot be instantiated directly. It requires
        subclassing and implementation of the abstract methods defined in the
        Worker class.

        The worker receives the same event instance and argument objects passed to
        :meth:`~edri.abstract.ManagerBase.start_worker`.
        EDRI does not copy them. Mutations made by the worker can therefore be
        visible to the caller; callers that require isolation must copy the input
        or construct a dedicated data-transfer event before starting the worker.
    """

    def __init__(self, pipe: Connection, event: T, name: str, *args: Any, **kwargs: Any) -> None:  # pragma: no cover
        # The positional mapping (pipe->event, event->resolver, name->args) does
        # not match the Worker dataclass field order, so the Worker dataclass fields
        # receive mismatched values (see mypy-abstract-report.md); left as-is per task rules.
        Worker.__init__(self, pipe, event, name, *args, **kwargs)
        Thread.__init__(self)
        # A thread cannot be force-terminated, so a still-running worker thread
        # would block the Manager process from exiting at shutdown. Marking it a
        # daemon lets the process exit; the Manager actively reaps only its
        # process-backed workers (see ManagerBase._shutdown_workers).
        self.daemon = True