hololinked.core.eventloop.scheduler.ThreadedScheduler

Bases: Scheduler

Scheduler class to schedule the operations of a thing in a threaded loop.

Source code in repo/hololinked/hololinked/core/eventloop/scheduler.py

class ThreadedScheduler(Scheduler):
    """Scheduler class to schedule the operations of a thing in a threaded loop."""

    def __init__(self, instance: Thing, eventloop: EventLoop) -> None:
        super().__init__(instance, eventloop)
        self._job = None
        self._execution_thread = None
        self._one_shot = True
        self._operation_execution_ready_event = CrossLoopEvent()
        self._operation_execution_complete_event = CrossLoopEvent()

    @property
    def has_job(self) -> bool:
        return self._job is not None

    @property
    def next_job(self) -> Job:
        if self._job is None:
            raise RuntimeError("No job to execute")
        return self._job

    def dispatch_job(self, job: Job) -> None:
        """
        Store the job and start a thread to execute it on the `Thing` instance.

        The drain half goes onto the `EventLoop`'s own asyncio loop, not the caller's - see `AsyncScheduler`.

        Parameters
        ----------
        job: Job
            the operation to run, and the future that answers whoever submitted it
        """
        self._job = job
        self.eventloop.run_coro_threadsafe(self.eventloop.tunnel_message_to_things(self))
        self._execution_thread = threading.Thread(
            target=asyncio.run,
            args=(self.eventloop.run_thing_instance(self.instance, self),),
        )
        self._execution_thread.start()
        self._job_queued_event.set()

Functions

__init__

__init__(instance: Thing, eventloop: EventLoop) -> None

Source code in repo/hololinked/hololinked/core/eventloop/scheduler.py

def __init__(self, instance: Thing, eventloop: EventLoop) -> None:
    super().__init__(instance, eventloop)
    self._job = None
    self._execution_thread = None
    self._one_shot = True
    self._operation_execution_ready_event = CrossLoopEvent()
    self._operation_execution_complete_event = CrossLoopEvent()

dispatch_job

dispatch_job(job: Job) -> None

Store the job and start a thread to execute it on the Thing instance.

The drain half goes onto the EventLoop's own asyncio loop, not the caller's - see AsyncScheduler.

Parameters:

Name	Type	Description	Default
job	Job	the operation to run, and the future that answers whoever submitted it	required

Source code in repo/hololinked/hololinked/core/eventloop/scheduler.py

def dispatch_job(self, job: Job) -> None:
    """
    Store the job and start a thread to execute it on the `Thing` instance.

    The drain half goes onto the `EventLoop`'s own asyncio loop, not the caller's - see `AsyncScheduler`.

    Parameters
    ----------
    job: Job
        the operation to run, and the future that answers whoever submitted it
    """
    self._job = job
    self.eventloop.run_coro_threadsafe(self.eventloop.tunnel_message_to_things(self))
    self._execution_thread = threading.Thread(
        target=asyncio.run,
        args=(self.eventloop.run_thing_instance(self.instance, self),),
    )
    self._execution_thread.start()
    self._job_queued_event.set()

next_job

next_job() -> Job

Source code in repo/hololinked/hololinked/core/eventloop/scheduler.py

@property
def next_job(self) -> Job:
    if self._job is None:
        raise RuntimeError("No job to execute")
    return self._job

has_job

has_job() -> bool

Source code in repo/hololinked/hololinked/core/eventloop/scheduler.py

@property
def has_job(self) -> bool:
    return self._job is not None