Add stop functionality

.vscode/settings.json

@@ -0,0 +1,6 @@
+{
+  "[python]": {
+    "editor.detectIndentation": false,
+    "editor.tabSize": 4
+  }
+}

backend/loader.py

@@ -1,4 +1,4 @@
-from asyncio import Queue
+from asyncio import Queue, get_event_loop, sleep, wait_for
 from json.decoder import JSONDecodeError
 from logging import getLogger
 from os import listdir, path
@@ -16,7 +16,7 @@ except UnsupportedLibc:
     from watchdog.observers.fsevents import FSEventsObserver as Observer
 
 from injector import get_tab, inject_to_tab
-from plugin import PluginWrapper
+from plugin_wrapper import PluginWrapper
 
 
 class FileChangeHandler(RegexMatchingEventHandler):
@@ -29,7 +29,7 @@ class FileChangeHandler(RegexMatchingEventHandler):
     def maybe_reload(self, src_path):
         plugin_dir = Path(path.relpath(src_path, self.plugin_path)).parts[0]
         if exists(path.join(self.plugin_path, plugin_dir, "plugin.json")):
-            self.queue.put_nowait((path.join(self.plugin_path, plugin_dir, "main.py"), plugin_dir, True))
+            self.queue.put_nowait(plugin_dir, True)
 
     def on_created(self, event):
         src_path = event.src_path
@@ -102,9 +102,9 @@ class Loader:
         with open(path.join(self.plugin_path, plugin.plugin_directory, "dist/index.js"), 'r') as bundle:
             return web.Response(text=bundle.read(), content_type="application/javascript")
 
-    def import_plugin(self, file, plugin_directory, refresh=False):
+    def import_plugin(self, plugin_directory, refresh=False):
         try:
-            plugin = PluginWrapper(file, plugin_directory, self.plugin_path)
+            plugin = PluginWrapper(plugin_directory, self.plugin_path)
             if plugin.name in self.plugins:
                     if not "debug" in plugin.flags and refresh:
                         self.logger.info(f"Plugin {plugin.name} is already loaded and has requested to not be re-loaded")
@@ -112,13 +112,12 @@ class Loader:
                     else:
                         self.plugins[plugin.name].stop()
                         self.plugins.pop(plugin.name, None)
-            if plugin.passive:
+            self.plugins[plugin.name] = plugin
-                self.logger.info(f"Plugin {plugin.name} is passive")
+            self.loop.create_task(plugin.start())
-            self.plugins[plugin.name] = plugin.start()
             self.logger.info(f"Loaded {plugin.name}")
             self.loop.create_task(self.dispatch_plugin(plugin.name))
         except Exception as e:
-            self.logger.error(f"Could not load {file}. {e}")
+            self.logger.error(f"Could not load {plugin_directory}. {e}")
             print_exc()
 
     async def dispatch_plugin(self, name):
@@ -130,7 +129,7 @@ class Loader:
         directories = [i for i in listdir(self.plugin_path) if path.isdir(path.join(self.plugin_path, i)) and path.isfile(path.join(self.plugin_path, i, "plugin.json"))]
         for directory in directories:
             self.logger.info(f"found plugin: {directory}")
-            self.import_plugin(path.join(self.plugin_path, directory, "main.py"), directory)
+            self.import_plugin(directory)
 
     async def handle_reloads(self):
         while True:
@@ -143,16 +142,15 @@ class Loader:
         method_name = request.match_info["method_name"]
         try:
             method_info = await request.json()
-            args = method_info["args"]
+            method_args = method_info["args"]
         except JSONDecodeError:
-            args = {}
+            method_args = {}
         try:
           if method_name.startswith("_"):
               raise RuntimeError("Tried to call private method")
-          res["result"] = await plugin.execute_method(method_name, args)
+          res = await plugin.call_method(method_name, method_args)
-          res["success"] = True
         except Exception as e:
-            res["result"] = str(e)
+            res["result"] = repr(e)
             res["success"] = False
         return web.json_response(res)
 

backend/plugin.py

@@ -1,115 +0,0 @@
-import multiprocessing
-from asyncio import (Lock, get_event_loop, new_event_loop,
-                     open_unix_connection, set_event_loop, sleep,
-                     start_unix_server)
-from concurrent.futures import ProcessPoolExecutor
-from importlib.util import module_from_spec, spec_from_file_location
-from json import dumps, load, loads
-from os import path, setuid
-from signal import SIGINT, signal
-from sys import exit
-from time import time
-
-multiprocessing.set_start_method("fork")
-
-class PluginWrapper:
-    def __init__(self, file, plugin_directory, plugin_path) -> None:
-        self.file = file
-        self.plugin_directory = plugin_directory
-        self.reader = None
-        self.writer = None
-        self.socket_addr = f"/tmp/plugin_socket_{time()}"
-        self.method_call_lock = Lock()
-
-        json = load(open(path.join(plugin_path, plugin_directory, "plugin.json"), "r"))
-
-        self.legacy = False
-        self.main_view_html = json["main_view_html"] if "main_view_html" in json else ""
-        self.tile_view_html = json["tile_view_html"] if "tile_view_html" in json else ""
-        self.legacy = self.main_view_html or self.tile_view_html
-
-        self.name = json["name"]
-        self.author = json["author"]
-        self.flags = json["flags"]
-
-        self.passive = not path.isfile(self.file)
-
-    def __str__(self) -> str:
-        return self.name
-
-    def _init(self):
-        signal(SIGINT, lambda s, f: exit(0))
-
-        set_event_loop(new_event_loop())
-        if self.passive:
-            return
-        setuid(0 if "root" in self.flags else 1000)
-        spec = spec_from_file_location("_", self.file)
-        module = module_from_spec(spec)
-        spec.loader.exec_module(module)
-        self.Plugin = module.Plugin
-
-        if hasattr(self.Plugin, "_main"):
-            get_event_loop().create_task(self.Plugin._main(self.Plugin))
-        get_event_loop().create_task(self._setup_socket())
-        get_event_loop().run_forever()
-
-    async def _setup_socket(self):
-        self.socket = await start_unix_server(self._listen_for_method_call, path=self.socket_addr)
-
-    async def _listen_for_method_call(self, reader, writer):
-        while True:
-            data = loads((await reader.readline()).decode("utf-8"))
-            if "stop" in data:
-                get_event_loop().stop()
-                while get_event_loop().is_running():
-                    await sleep(0)
-                get_event_loop().close()
-                return
-            d = {"res": None, "success": True}
-            try:
-                d["res"] = await getattr(self.Plugin, data["method"])(self.Plugin, **data["args"])
-            except Exception as e:
-                d["res"] = str(e)
-                d["success"] = False
-            finally:
-                writer.write((dumps(d)+"\n").encode("utf-8"))
-                await writer.drain()
-
-    async def _open_socket_if_not_exists(self):
-        if not self.reader:
-            while True:
-                try:
-                    self.reader, self.writer = await open_unix_connection(self.socket_addr)
-                    break
-                except:
-                    await sleep(0)
-
-    def start(self):
-        if self.passive:
-            return self
-        multiprocessing.Process(target=self._init).start()
-        return self
-
-    def stop(self):
-        if self.passive:
-            return
-        async def _(self):
-            await self._open_socket_if_not_exists()
-            self.writer.write((dumps({"stop": True})+"\n").encode("utf-8"))
-            await self.writer.drain()
-            self.writer.close()
-        get_event_loop().create_task(_(self))
-
-    async def execute_method(self, method_name, kwargs):
-        if self.passive:
-            raise RuntimeError("This plugin is passive (aka does not implement main.py)")
-        async with self.method_call_lock:
-            await self._open_socket_if_not_exists()
-            self.writer.write(
-                (dumps({"method": method_name, "args": kwargs})+"\n").encode("utf-8"))
-            await self.writer.drain()
-            res = loads((await self.reader.readline()).decode("utf-8"))
-            if not res["success"]:
-                raise Exception(res["res"])
-            return res["res"]

backend/plugin/binary_plugin.py

@@ -0,0 +1,59 @@
+import os
+from asyncio import get_event_loop, sleep, subprocess
+from posixpath import join
+from tempfile import mkdtemp
+
+from plugin_protocol import PluginProtocolServer
+
+
+class BinaryPlugin:
+    def __init__(self, plugin_directory, file_name, flags, logger) -> None:
+        self.server = PluginProtocolServer(self)
+        self.connection = None
+        self.process = None
+
+        self.flags = flags
+        self.logger = logger
+
+        self.plugin_directory = plugin_directory
+        self.file_name = file_name
+
+
+    async def start(self):
+        if self.connection and self.connection.is_serving:
+            self.connection.close()
+
+        self.unix_socket_path = BinaryPlugin.generate_socket_path()
+        self.logger.debug(f"starting unix server on {self.unix_socket_path}")
+        self.connection = await get_event_loop().create_unix_server(lambda: self.server, path=self.unix_socket_path)
+
+        env = dict(DECKY_PLUGIN_SOCKET = self.unix_socket_path)
+        self.process = await subprocess.create_subprocess_exec(join(self.plugin_directory, self.file_name), env=env)
+        get_event_loop().create_task(self.process_loop())
+
+    async def stop(self):
+        self.stopped = True
+        if self.connection and self.connection.is_serving:
+            self.connection.close()
+
+        if self.process and self.process.is_alive:
+            self.process.terminate()
+
+    async def process_loop(self):
+        await self.process.wait()
+        if not self.stopped:
+            self.logger.info("backend process was killed - restarting in 10 seconds")
+            await sleep(10)
+            await self.start()
+
+    def generate_socket_path():
+        tmp_dir = mkdtemp("decky-plugin")
+        os.chown(tmp_dir, 1000, 1000)
+        return join(tmp_dir, "socket")
+
+    # called on the server/loader process
+    async def call_method(self, method_name, method_args):
+        if self.process.returncode == None:
+            return dict(success = False, result = "Process not alive")
+
+        return await self.server.call_method(method_name, method_args)

backend/plugin/passive_plugin.py

@@ -0,0 +1,18 @@
+class PassivePlugin:
+    def __init__(self, logger) -> None:
+        self.logger
+        pass
+
+    def call_method(self, method_name, args):
+        self.logger.debug(f"Tried to call method {method_name}, but plugin is in passive mode")
+        pass
+
+    def execute_method(self, method_name, method_args):
+        self.logger.debug(f"Tried to execute method {method_name}, but plugin is in passive mode")
+        pass
+
+    async def start(self):
+        pass
+
+    async def stop(self):
+        pass

backend/plugin/plugin.py

@@ -0,0 +1,18 @@
+from posixpath import join
+
+from genericpath import isfile
+
+from plugin.binary_plugin import BinaryPlugin
+from plugin.passive_plugin import PassivePlugin
+from plugin.python_plugin import PythonPlugin
+
+
+def get_plugin_backend(spec, plugin_directory, flags, logger):
+    if spec == None and isfile(join(plugin_directory, "main.py")):
+        return PythonPlugin(plugin_directory, "main.py", flags, logger)
+    elif spec["type"] == "python":
+        return PythonPlugin(plugin_directory, spec["file"], flags, logger)
+    elif spec["type"] == "binary":
+        return BinaryPlugin(plugin_directory, spec["file"], flags, logger)
+    else:
+        return PassivePlugin(logger)

backend/plugin/python_plugin.py

@@ -0,0 +1,129 @@
+import json
+import multiprocessing
+import os
+import uuid
+from asyncio import (Protocol, get_event_loop, new_event_loop, set_event_loop,
+                     sleep)
+from importlib.util import module_from_spec, spec_from_file_location
+from posixpath import join
+from signal import SIGINT, signal
+from tempfile import mkdtemp
+
+from plugin_protocol import PluginProtocolServer
+
+multiprocessing.set_start_method("fork")
+
+# only useable by the python backend
+class PluginProtocolClient(Protocol):
+    def __init__(self, backend, logger) -> None:
+        super().__init__()
+        self.backend = backend
+        self.logger = logger
+
+    def connection_made(self, transport):
+        self.transport = transport
+
+    def data_received(self, data: bytes) -> None:
+        message = json.loads(data.decode("utf-8"))
+        message_id = str(uuid.UUID(message["id"]))
+        message_type = message["type"]
+        payload = message["payload"]
+
+        self.logger.debug(f"received {message_id} {message_type} {payload}")
+        if message_type == "method_call":
+          get_event_loop().create_task(self.handle_method_call(message_id, payload["name"], payload["args"]))
+
+    async def handle_method_call(self, message_id, method_name, method_args):
+        try:
+            result = await self.backend.execute_method(method_name, method_args)
+            self.respond_message(message_id, "method_response", dict(success = True, result = result))
+        except AttributeError as e:
+            self.respond_message(message_id, "method_response", dict(success = False, result = f"plugin does not expose a method called {method_name}"))
+        except Exception as e:
+            self.respond_message(message_id, "method_response", dict(success = False, result = str(e)))
+
+    def respond_message(self, message_id, message_type, payload):
+        self.logger.debug(f"sending {message_id} {message_type} {payload}")
+        message = json.dumps(dict(id = str(message_id), type = message_type, payload = payload))
+        self.transport.write(message.encode('utf-8'))
+
+
+class PythonPlugin:
+    def __init__(self, plugin_directory, file_name, flags, logger) -> None:
+        self.client = PluginProtocolClient(self, logger)
+        self.server = PluginProtocolServer(self)
+        self.connection = None
+        self.process = None
+        self.stopped = False
+
+        self.plugin_directory = plugin_directory
+        self.file_name = file_name
+        self.flags = flags
+        self.logger = logger
+
+    def _init(self):
+        self.logger.debug(f"child process Initializing")
+        signal(SIGINT, lambda s, f: exit(0))
+
+        set_event_loop(new_event_loop())
+        # TODO: both processes can access the socket
+        # setuid(0 if "root" in self.flags else 1000)
+        spec = spec_from_file_location("_", join(self.plugin_directory, self.file_name))
+        module = module_from_spec(spec)
+        spec.loader.exec_module(module)
+        self.Plugin = module.Plugin
+
+        if hasattr(self.Plugin, "_main"):
+            self.logger.debug("Found _main, calling it")
+            get_event_loop().create_task(self.Plugin._main(self.Plugin))
+
+        get_event_loop().create_task(self._connect())
+        get_event_loop().run_forever()
+
+    async def _connect(self):
+        self.logger.debug(f"connecting to unix server on {self.unix_socket_path}")
+        await get_event_loop().create_unix_connection(lambda: self.client, path=self.unix_socket_path)
+
+    async def start(self):
+        if self.connection:
+            self.connection.close()
+
+        self.unix_socket_path = PythonPlugin.generate_socket_path()
+        self.logger.debug(f"starting unix server on {self.unix_socket_path}")
+        self.connection = await get_event_loop().create_unix_server(lambda: self.server, path=self.unix_socket_path)
+
+        self.process = multiprocessing.Process(target=self._init)
+        self.process.start()
+        get_event_loop().create_task(self.process_loop())
+        self.stopped = False
+
+    async def stop(self):
+        self.stopped = True
+        if self.connection:
+            self.connection.close()
+
+        if self.process and self.process.is_alive:
+            self.process.terminate()
+
+    async def process_loop(self):
+        await get_event_loop().run_in_executor(None, self.process.join)
+        if not self.stopped:
+            self.logger.info("backend process was killed - restarting in 10 seconds")
+            await sleep(10)
+            await self.start()
+
+    # called on the server/loader process
+    async def call_method(self, method_name, method_args):
+        if not self.process.is_alive():
+            return dict(success = False, result = "Process not alive")
+
+        return await self.server.call_method(method_name, method_args)
+
+    # called on the client
+    def execute_method(self, method_name, method_args):
+        return getattr(self.Plugin, method_name)(self.Plugin, **method_args)
+
+    def generate_socket_path():
+        tmp_dir = mkdtemp("decky-plugin")
+        os.chown(tmp_dir, 1000, 1000)
+        return join(tmp_dir, "socket")

backend/plugin_protocol.py

@@ -0,0 +1,46 @@
+import json
+import uuid
+from asyncio import Protocol, TimeoutError, get_event_loop, wait_for
+from gc import callbacks
+from subprocess import call
+
+
+class PluginProtocolServer(Protocol):
+  def __init__(self, backend) -> None:
+    super().__init__()
+    self.backend = backend
+    self.callbacks = {}
+
+  def connection_made(self, transport):
+    self.transport = transport
+
+  def data_received(self, data: bytes) -> None:
+    message = json.loads(data.decode("utf-8"))
+    message_id = str(uuid.UUID(message["id"]))
+    message_type = message["type"]
+    payload = message["payload"]
+
+    if message_type == "method_response":
+      get_event_loop().create_task(self.handle_method_response(message_id, payload["success"], payload["result"]))
+
+  async def handle_method_response(self, message_id, success, result):
+    if message_id in self.callbacks:
+      self.callbacks[message_id].set_result(dict(success = success, result = result))
+      del self.callbacks[message_id]
+
+  async def send_message(self, type, payload):
+    id = str(uuid.uuid4())
+    callback = get_event_loop().create_future()
+    message = json.dumps(dict(id = id, type = type, payload = payload))
+
+    self.callbacks[id] = callback
+    self.transport.write(message.encode('utf-8'))
+
+    try:
+      return await wait_for(callback, 10)
+    except TimeoutError as e:
+      del self.callbacks[id]
+      raise e
+
+  def call_method(self, method_name, method_args):
+    return self.send_message("method_call", dict(name = method_name, args = method_args))

backend/plugin_wrapper.py

@@ -0,0 +1,37 @@
+import multiprocessing
+from json import load
+from logging import getLogger
+from os import path
+
+from plugin.plugin import get_plugin_backend
+
+
+class PluginWrapper:
+    def __init__(self, plugin_relative_directory, plugin_path) -> None:
+        self.plugin_directory = path.join(plugin_path, plugin_relative_directory)
+
+        json = load(open(path.join(self.plugin_directory, "plugin.json"), "r"))
+
+        self.legacy = False
+        self.main_view_html = json["main_view_html"] if "main_view_html" in json else ""
+        self.tile_view_html = json["tile_view_html"] if "tile_view_html" in json else ""
+        self.legacy = self.main_view_html or self.tile_view_html
+
+        self.name = json["name"]
+        self.author = json["author"]
+        self.flags = json["flags"]
+        self.logger = getLogger(f"{self.name}")
+
+        self.backend = get_plugin_backend(json.get("backend"), self.plugin_directory, self.flags, self.logger)
+
+    def call_method(self, method_name, args):
+        return self.backend.call_method(method_name, args)
+
+    def start(self):
+        return self.backend.start()
+
+    def stop(self):
+        return self.backend.stop()
+
+    def __str__(self) -> str:
+        return self.name