Skip to content

hololinked.core.eventloop.scheduler.Scheduler

Scheduler class to schedule the operations of a thing either in queued mode, or a one-shot mode in either async or threaded loops.

UML Diagram subclasses

Source code in repo/hololinked/hololinked/core/eventloop/scheduler.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
class Scheduler:
    """
    Scheduler class to schedule the operations of a thing either in queued mode, or a one-shot mode in either async or threaded loops.

    [UML Diagram subclasses](http://docs.hololinked.dev/UML/PDF/Scheduler.pdf)
    """

    _operation_execution_complete_event: CrossLoopEvent
    _operation_execution_ready_event: CrossLoopEvent

    def __init__(self, instance: Thing, eventloop: EventLoop) -> None:
        self.instance = instance  # type: Thing
        self.eventloop = eventloop  # type: EventLoop
        self.run = True  # type: bool
        self._one_shot = False  # type: bool
        self._last_operation_request = Undefined  # type: Operation
        self._last_operation_reply = Undefined  # type: Reply
        self._job_queued_event = CrossLoopEvent()  # type: CrossLoopEvent

    @property
    def last_operation_request(self) -> Operation:
        return self._last_operation_request

    @last_operation_request.setter
    def last_operation_request(self, value: Operation):
        self._last_operation_request = value
        self._operation_execution_ready_event.set()

    def reset_operation_request(self) -> None:
        self._last_operation_request = Undefined

    @property
    def last_operation_reply(self) -> Reply:
        return self._last_operation_reply

    @last_operation_reply.setter
    def last_operation_reply(self, value: Reply):
        self._last_operation_request = Undefined
        self._last_operation_reply = value
        self._operation_execution_complete_event.set()
        if self._one_shot:
            self.run = False

    def reset_operation_reply(self) -> None:
        self._last_operation_reply = Undefined

    async def wait_for_job(self) -> None:
        await self._job_queued_event.wait()
        self._job_queued_event.clear()

    async def wait_for_operation(self) -> None:
        """Wait, on the `Thing`'s loop, until an operation has been handed over for execution."""
        await self._operation_execution_ready_event.wait()
        self._operation_execution_ready_event.clear()

    async def wait_for_reply(self) -> None:
        """Wait, on the listener loop, until the `Thing` has finished and produced a reply."""
        await self._operation_execution_complete_event.wait()
        self._operation_execution_complete_event.clear()

    @property
    def has_job(self) -> bool:
        raise NotImplementedError("has_job method must be implemented in the subclass")

    @property
    def next_job(self) -> Job:
        raise NotImplementedError("next_job method must be implemented in the subclass")

    def dispatch_job(self, job: Job) -> None:
        raise NotImplementedError("dispatch_job method must be implemented in the subclass")

    def cleanup(self):
        self.run = False
        self._job_queued_event.set()
        self._operation_execution_ready_event.set()
        self._operation_execution_complete_event.set()

    @classmethod
    def format_reply_tuple(self, return_value: Any) -> Reply:
        raise NotImplementedError("Implement format_reply_tuple in subclass")

Functions

__init__ 

__init__(instance: Thing, eventloop: EventLoop) -> None
Source code in repo/hololinked/hololinked/core/eventloop/scheduler.py 
46
47
48
49
50
51
52
53
def __init__(self, instance: Thing, eventloop: EventLoop) -> None:
    self.instance = instance  # type: Thing
    self.eventloop = eventloop  # type: EventLoop
    self.run = True  # type: bool
    self._one_shot = False  # type: bool
    self._last_operation_request = Undefined  # type: Operation
    self._last_operation_reply = Undefined  # type: Reply
    self._job_queued_event = CrossLoopEvent()  # type: CrossLoopEvent

dispatch_job 

dispatch_job(job: Job) -> None
Source code in repo/hololinked/hololinked/core/eventloop/scheduler.py 
104
105
def dispatch_job(self, job: Job) -> None:
    raise NotImplementedError("dispatch_job method must be implemented in the subclass")

next_job 

next_job() -> Job
Source code in repo/hololinked/hololinked/core/eventloop/scheduler.py 
100
101
102
@property
def next_job(self) -> Job:
    raise NotImplementedError("next_job method must be implemented in the subclass")

has_job 

has_job() -> bool
Source code in repo/hololinked/hololinked/core/eventloop/scheduler.py 
96
97
98
@property
def has_job(self) -> bool:
    raise NotImplementedError("has_job method must be implemented in the subclass")

wait_for_job async 

wait_for_job() -> None
Source code in repo/hololinked/hololinked/core/eventloop/scheduler.py 
82
83
84
async def wait_for_job(self) -> None:
    await self._job_queued_event.wait()
    self._job_queued_event.clear()

wait_for_operation async 

wait_for_operation() -> None

Wait, on the Thing's loop, until an operation has been handed over for execution.

Source code in repo/hololinked/hololinked/core/eventloop/scheduler.py 
86
87
88
89
async def wait_for_operation(self) -> None:
    """Wait, on the `Thing`'s loop, until an operation has been handed over for execution."""
    await self._operation_execution_ready_event.wait()
    self._operation_execution_ready_event.clear()

wait_for_reply async 

wait_for_reply() -> None

Wait, on the listener loop, until the Thing has finished and produced a reply.

Source code in repo/hololinked/hololinked/core/eventloop/scheduler.py 
91
92
93
94
async def wait_for_reply(self) -> None:
    """Wait, on the listener loop, until the `Thing` has finished and produced a reply."""
    await self._operation_execution_complete_event.wait()
    self._operation_execution_complete_event.clear()

last_operation_request 

last_operation_request(value: Operation)
Source code in repo/hololinked/hololinked/core/eventloop/scheduler.py 
59
60
61
62
@last_operation_request.setter
def last_operation_request(self, value: Operation):
    self._last_operation_request = value
    self._operation_execution_ready_event.set()

last_operation_reply 

last_operation_reply(value: Reply)
Source code in repo/hololinked/hololinked/core/eventloop/scheduler.py 
71
72
73
74
75
76
77
@last_operation_reply.setter
def last_operation_reply(self, value: Reply):
    self._last_operation_request = Undefined
    self._last_operation_reply = value
    self._operation_execution_complete_event.set()
    if self._one_shot:
        self.run = False

reset_operation_request 

reset_operation_request() -> None
Source code in repo/hololinked/hololinked/core/eventloop/scheduler.py 
64
65
def reset_operation_request(self) -> None:
    self._last_operation_request = Undefined

reset_operation_reply 

reset_operation_reply() -> None
Source code in repo/hololinked/hololinked/core/eventloop/scheduler.py 
79
80
def reset_operation_reply(self) -> None:
    self._last_operation_reply = Undefined

format_reply_tuple classmethod 

format_reply_tuple(return_value: Any) -> Reply
Source code in repo/hololinked/hololinked/core/eventloop/scheduler.py 
113
114
115
@classmethod
def format_reply_tuple(self, return_value: Any) -> Reply:
    raise NotImplementedError("Implement format_reply_tuple in subclass")

cleanup 

cleanup()
Source code in repo/hololinked/hololinked/core/eventloop/scheduler.py 
107
108
109
110
111
def cleanup(self):
    self.run = False
    self._job_queued_event.set()
    self._operation_execution_ready_event.set()
    self._operation_execution_complete_event.set()