Mirrors/chatmail-core
1
0
Fork 0
mirror of https://github.com/chatmail/core.git synced 2026-05-09 01:46:30 +03:00
Code Issues Packages Projects Releases Wiki Activity

refactor: strike events in rpc-client request handling, get result from queue

Browse Source
This commit is contained in:
holger krekel
2025-11-23 20:15:21 +01:00
parent 6077499f07
commit 8eef79f95d
1 changed files with 13 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
deltachat-rpc-client/src/deltachat_rpc_client/rpc.py
View File

@@ -9,7 +9,7 @@ import os
import subprocess import subprocess
import sys import sys
from queue import Empty, Queue from queue import Empty, Queue
from threading import Event, Thread from threading import Thread
from typing import Any, Iterator, Optional from typing import Any, Iterator, Optional
@@ -17,25 +17,6 @@ class JsonRpcError(Exception):
"""JSON-RPC error.""" """JSON-RPC error."""
class RpcFuture:
"""RPC future waiting for RPC call result."""
def __init__(self, rpc: "Rpc", request_id: int, event: Event):
self.rpc = rpc
self.request_id = request_id
self.event = event
def __call__(self):
"""Wait for the future to return the result."""
self.event.wait()
response = self.rpc.request_results.pop(self.request_id)
if "error" in response:
raise JsonRpcError(response["error"])
if "result" in response:
return response["result"]
return None
class RpcMethod: class RpcMethod:
"""RPC method.""" """RPC method."""
@@ -57,11 +38,17 @@ class RpcMethod:
"params": args, "params": args,
"id": request_id, "id": request_id,
} }
event = Event() self.rpc.request_results[request_id] = queue = Queue()
self.rpc.request_events[request_id] = event
self.rpc.request_queue.put(request) self.rpc.request_queue.put(request)
return RpcFuture(self.rpc, request_id, event) def rpc_future():
"""Wait for the request to receive a result."""
response = queue.get()
if "error" in response:
raise JsonRpcError(response["error"])
return response.get("result", None)
return rpc_future
class Rpc: class Rpc:
@@ -82,10 +69,8 @@ class Rpc:
self.process: subprocess.Popen self.process: subprocess.Popen
self.id_iterator: Iterator[int] self.id_iterator: Iterator[int]
self.event_queues: dict[int, Queue] self.event_queues: dict[int, Queue]
# Map from request ID to `threading.Event`. # Map from request ID to a Queue which provides a single result
self.request_events: dict[int, Event] self.request_results: dict[int, Queue]
# Map from request ID to the result.
self.request_results: dict[int, Any]
self.request_queue: Queue[Any] self.request_queue: Queue[Any]
self.closing: bool self.closing: bool
self.reader_thread: Thread self.reader_thread: Thread
@@ -114,7 +99,6 @@ class Rpc:
) )
self.id_iterator = itertools.count(start=1) self.id_iterator = itertools.count(start=1)
self.event_queues = {} self.event_queues = {}
self.request_events = {}
self.request_results = {} self.request_results = {}
self.request_queue = Queue() self.request_queue = Queue()
self.closing = False self.closing = False
@@ -149,9 +133,7 @@ class Rpc:
response = json.loads(line) response = json.loads(line)
if "id" in response: if "id" in response:
response_id = response["id"] response_id = response["id"]
event = self.request_events.pop(response_id) self.request_results.pop(response_id).put(response)
self.request_results[response_id] = response
event.set()
else: else:
logging.warning("Got a response without ID: %s", response) logging.warning("Got a response without ID: %s", response)
except Exception: except Exception: