threads.thread_controlモジュール: pause/resumeのイベント処理の実装

[Geson-anko]

タスク内容

#35 に、Pause/Resumeが実行された際のイベント呼び出し処理を実装します。

実装

• ControllerCommandHandlerのstop_if_pauseに実装します。
• Pause/Resumeイベントのインターフェイスを注入できるMixinクラスも定義します。

スケッチ

イベント処理

from typing import Callable

type OnPausedCallbackType = Callable[[], None]
type OnResumedCallbackType = Callable[[], None]

class ControllerCommandHandler:
    def __init__(
        self, 
        controller: ReadOnlyController,
        on_paused_callback: OnPausedCallbackType = lambda : None,
        on_resumed_callback: OnResumedCallbackType = lambda: None,
    ):
        self._controller = controller
        self.on_paused = on_paused_callback
        self.on_resumed = on_resumed_callback

    def stop_if_pause(self, check_resume_interval: float = 1.0) -> None:

        paused = False
        if self._controller.is_pause():
            self.on_paused()
            paused = True

        while not self._controller.wait_for_resume(check_resume_interval) and self._controller.is_active():
            #  whileの比較順序は変更しないこと (`is_paused`と`wait_for_resume`の実行間隔を最小化。 )
            pass

        if paused:
            self.on_resumed()

Mixin

class ThreadEventMixin:
    def on_paused(self):
        pass
    def on_resumed(self):
        pass

達成条件

• イベント呼び出し処理の実装
• テスト
• Mixinの実装