mirror of
https://github.com/SteamDeckHomebrew/decky-loader.git
synced 2026-08-31 12:37:48 +00:00
Lukas Senionis
45 lines
1.2 KiB
Python
45 lines
1.2 KiB
Python
from typing import Any, TypedDict
|
|
from enum import IntEnum
|
|
from uuid import uuid4
|
|
from asyncio import Event
|
|
|
|
class SocketMessageType(IntEnum):
|
|
CALL = 0
|
|
RESPONSE = 1
|
|
EVENT = 2
|
|
|
|
class SocketResponseDict(TypedDict):
|
|
type: SocketMessageType
|
|
id: str
|
|
success: bool
|
|
res: Any
|
|
|
|
class MethodCallResponse:
|
|
def __init__(self, success: bool, result: Any) -> None:
|
|
self.success = success
|
|
self.result = result
|
|
|
|
class PluginStopped(Exception):
|
|
pass
|
|
|
|
class MethodCallRequest:
|
|
def __init__(self) -> None:
|
|
self.id = str(uuid4())
|
|
self.event = Event()
|
|
self.response: MethodCallResponse | PluginStopped
|
|
|
|
def set_result(self, dc: SocketResponseDict):
|
|
self.response = MethodCallResponse(dc["success"], dc["res"])
|
|
self.event.set()
|
|
|
|
def cancel(self):
|
|
self.response = PluginStopped("Plugin has been stopped")
|
|
self.event.set()
|
|
|
|
async def wait_for_result(self):
|
|
await self.event.wait()
|
|
if isinstance(self.response, PluginStopped):
|
|
raise self.response
|
|
if not self.response.success:
|
|
raise Exception(self.response.result)
|
|
return self.response.result |