Newton_Editor/plugins/lsp_client/pylspclient/lsp_endpoint.py

113 lines
3.9 KiB
Python
Raw Normal View History

2023-11-04 19:38:20 +00:00
from __future__ import print_function
import threading
from . import lsp_structs
class LspEndpoint(threading.Thread):
def __init__(self, json_rpc_endpoint, method_callbacks = {}, notify_callbacks = {}, timeout = 2):
threading.Thread.__init__(self)
self.json_rpc_endpoint = json_rpc_endpoint
self.notify_callbacks = notify_callbacks
self.method_callbacks = method_callbacks
self.event_dict = {}
self.response_dict = {}
self.next_id = 0
self._timeout = timeout
self.shutdown_flag = False
def handle_result(self, rpc_id, result, error):
self.response_dict[rpc_id] = (result, error)
cond = self.event_dict[rpc_id]
cond.acquire()
cond.notify()
cond.release()
def stop(self):
self.shutdown_flag = True
def run(self):
while not self.shutdown_flag:
try:
jsonrpc_message = self.json_rpc_endpoint.recv_response()
if jsonrpc_message is None:
print("LSP Server Shutting Down...")
break
method = jsonrpc_message.get("method")
result = jsonrpc_message.get("result")
error = jsonrpc_message.get("error")
rpc_id = jsonrpc_message.get("id")
params = jsonrpc_message.get("params")
if method:
if rpc_id:
if method not in self.method_callbacks:
raise lsp_structs.ResponseError(
lsp_structs.ErrorCodes.MethodNotFound,
"Method not found: {method}"
.format(method=method))
result = self.method_callbacks[method](params)
self.send_response(rpc_id, result, None)
else:
if method not in self.notify_callbacks:
print("Notify method not found: {method}.".format(method=method))
else:
self.notify_callbacks[method](params)
else:
self.handle_result(rpc_id, result, error)
except lsp_structs.ResponseError as e:
self.send_response(rpc_id, None, e)
def send_response(self, id, result, error):
message_dict = {}
message_dict["jsonrpc"] = "2.0"
message_dict["id"] = id
if result:
message_dict["result"] = result
if error:
message_dict["error"] = error
self.json_rpc_endpoint.send_request(message_dict)
def send_message(self, method_name, params, id=None):
message_dict = {}
message_dict["jsonrpc"] = "2.0"
if id is not None:
message_dict["id"] = id
message_dict["method"] = method_name
message_dict["params"] = params
self.json_rpc_endpoint.send_request(message_dict)
def call_method(self, method_name, **kwargs):
current_id = self.next_id
self.next_id += 1
cond = threading.Condition()
self.event_dict[current_id] = cond
cond.acquire()
self.send_message(method_name, kwargs, current_id)
if self.shutdown_flag:
return None
if not cond.wait(timeout=self._timeout):
raise TimeoutError()
cond.release()
self.event_dict.pop(current_id)
result, error = self.response_dict.pop(current_id)
if error:
raise lsp_structs.ResponseError(error.get("code"),
error.get("message"),
error.get("data"))
return result
def send_notification(self, method_name, **kwargs):
self.send_message(method_name, kwargs)