Python によるスレッド版command pattern (車輪の再開発)

動機

pyOpenGLがメインスレッドでなくても動きそうなので、
バックエンドとしてスレッドで動くcommandパターンがないかなと思って、
探したけど、実用的なのがすぐ見つからなかったので、作りました。
たぶん、だれか作ってると思うんだけど。

実装

Woker.py

from __future__ import annotations
from abc import ABC, abstractmethod
from threading import Thread
from queue import Queue


class Command(ABC):
    """
    The Command interface declares a method for executing a command.
    """

    @abstractmethod
    def execute(self) -> None:
        pass

class QuitCommand(Command):
    def execute(self) -> None:
        pass

class CallbackCommand(Command):
    def __init__(self, func, *args, **kwargs):
        self.func = func
        self.args = args
        self.kwargs = kwargs
    def execute(self):
        self.func(*self.args, **self.kwargs)
    

class Worker:
    def __init__(self):
        self.queue = Queue()
        self.is_done = False
        self.thread = Thread(target=self.run)
    
    def run(self):
        while True:
            item = self.queue.get()
            if isinstance(item, QuitCommand):
                break
            item.execute()
        self.is_done = True
        
    
    def start(self):
        self.thread.start()
    
    def join(self):
        self.thread.join()
    
    def put(self, cmd):
        self.queue.put(cmd)
    
    def putQuitCommand(self):
        self.put(QuitCommand())
    
    def putCallback(self, func, *args, **kwargs):
        self.put(CallbackCommand(func, *args, **kwargs))
        
        
if __name__ == "__main__":
    import time

    w = Worker()
    w.start()

    w.putCallback(print, "hogehoge")
    w.putCallback(time.sleep, 1)
    w.putCallback(print, "fugafuga")
    w.putQuitCommand()

    w.join()

コメント

本来は、Commandクラスを色々派生させて、commandは作るべきだろうけど、
実用上実行したい関数をコールバックしてもらうほうが多いかなと思って、
コールバック出来るものを追加。

Quitのやり方はもっといい方法ありそうだけど、一旦は今のやり方でやってます。
また見直すかも。