itdominator
/
LSP_Manager
generated from itdominator/Python-With-Gtk-Template
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Initial wiring of client calls and responses

This commit is contained in:
itdominator 2024-09-14 01:16:47 -05:00
parent 13b126ef6e
commit e0664123da
9 changed files with 294 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/app.py
View File

@ -7,6 +7,7 @@ import os
# Application imports
from libs.debugging import debug_signal_handler
from libs.ipc_server import IPCServer
from libs.lsp_endpoint_server import LSPEndpointServer
from core.window import Window
@ -24,6 +25,7 @@ class Application:
if not settings_manager.is_trace_debug():
self.load_ipc(args, unknownargs)
self.load_lsp_server()
self.setup_debug_hook()
Window(args, unknownargs).main()
@ -41,6 +43,13 @@ class Application:
raise AppLaunchException(f"{APP_NAME} IPC Server Exists: Have sent path(s) to it and closing...")
def load_lsp_server(self):
lsp_server = LSPEndpointServer()
self.ipc_realization_check(lsp_server)
if not lsp_server.is_ipc_alive:
raise AppLaunchException(f"{APP_NAME} IPC Server Already Exists...")
def ipc_realization_check(self, ipc_server):
try:
ipc_server.create_ipc_listener()

1
src/core/controllers/base_controller.py
View File

@ -67,4 +67,3 @@ class BaseController(IPCSignalsMixin, KeyboardSignalsMixin, BaseControllerData):
self.base_container = BaseContainer()
settings_manager.register_signals_to_builder([self, self.base_container])

67
src/core/controllers/lsp_controller.py
View File

@ -3,13 +3,16 @@ import os
import signal
import subprocess
import threading
import json
# Lib imports
from gi.repository import GLib
# Application imports
from libs.dto.lsp_messages import LEN_HEADER, TYPE_HEADER, get_message_str, get_message_obj, definition_query, references_query, symbols_query
from libs.dto.lsp_message_structs import ClientRequest, ClientNotification, LSPResponse
from libs.dto.lsp_message_structs import \
LSPResponseTypes, ClientRequest, ClientNotification, LSPResponseRequest, LSPResponseNotification
from .lsp_controller_events import LSPControllerEvents
@ -22,7 +25,7 @@ def _log_list():
class LSPController:
class LSPController(LSPControllerEvents):
def __init__(self):
super(LSPController).__init__()
@ -45,8 +48,15 @@ class LSPController:
def _subscribe_to_events(self):
event_system.subscribe("client-send-request", self._client_send_request)
event_system.subscribe("client-send-notification", self._client_send_notification)
# event_system.subscribe("client-send-request", self._client_send_request)
# event_system.subscribe("client-send-notification", self._client_send_notification)
event_system.subscribe("textDocument/didOpen", self._lsp_did_open)
# event_system.subscribe("textDocument/didSave", self._lsp_did_save)
# event_system.subscribe("textDocument/didClose", self._lsp_did_close)
event_system.subscribe("textDocument/didChange", self._lsp_did_change)
event_system.subscribe("textDocument/definition", self._lsp_goto)
event_system.subscribe("textDocument/completion", self._lsp_completion)
def _client_send_request(self, method: str, uri: str, line: int, character: int):
if not method: return
@ -148,7 +158,8 @@ class LSPController:
except Exception as e:
self.log_list.add_log_entry(
"LSP Client Error",
LSPResponse(
LSPResponseRequest(
"2.0",
None,
{
"error": repr(e)
@ -207,41 +218,23 @@ class LSPController:
if not message_size: return
data = self.lsp_process.stdout.read(message_size)
jsonrpc_res = data.decode("utf-8")
lsp_response = LSPResponse(**get_message_obj(jsonrpc_res))
_data = self.lsp_process.stdout.read(message_size)
data = _data.decode("utf-8")
message = get_message_obj(data)
keys = message.keys()
lsp_response = None
if "result" in keys:
lsp_response = LSPResponseRequest(**get_message_obj(data))
if "method" in keys:
lsp_response = LSPResponseNotification(**get_message_obj(data))
response_id = -1
if not lsp_response: return
GLib.idle_add(self.handle_lsp_response, lsp_response)
def handle_lsp_response(self, lsp_response: LSPResponse):
method = self.request_list[lsp_response.id]
result = lsp_response.result
def handle_lsp_response(self, lsp_response: LSPResponseTypes):
self.log_list.add_log_entry("LSP Response", lsp_response)
self.process_lsp_response(method, result)
def process_lsp_response(self, method, result):
if isinstance(result, dict):
keys = result.keys()
return
if "error" in keys:
error = result["error"]
logger.debug(f"LSP Error Code: {error['code']}")
logger.debug(f"LSP Error Message:\n{error['message']}")
return
if "result" in keys:
result = result["result"]
if isinstance(result, dict):
keys = result.keys()
if "capabilities" in keys:
...
if isinstance(result, list):
...
if isinstance(result, tuple):
...
event_system.emit("respond-to-client", (get_message_str(lsp_response),))

59
src/core/controllers/lsp_controller_events.py Normal file
View File

@ -0,0 +1,59 @@
# Python imports
# Lib imports
from gi.repository import GLib
# Application imports
# from libs.dto.lsp_messages import LEN_HEADER, TYPE_HEADER, get_message_str, get_message_obj, definition_query, references_query, symbols_query
from libs.dto.lsp_messages import didopen_notification
class LSPControllerEvents:
def _lsp_did_open(self, data: dict):
method = data["method"]
params = didopen_notification["params"]
params["textDocument"]["uri"] = data["uri"]
params["textDocument"]["languageId"] = data["language_id"]
params["textDocument"]["text"] = data["text"]
GLib.idle_add( self.send_notification, method, params )
def _lsp_did_save(self, data: dict):
# self.send_notification(method, params)
...
def _lsp_did_close(self, data: dict):
# self.send_notification(method, params)
...
def _lsp_did_change(self, data: dict):
method = data["method"]
language_id = data["language_id"]
uri = data["uri"]
text = data["text"]
line = data["line"]
column = data["column"]
self.send_notification(method, params)
# return "{'notification':'some kind of response'}"
def _lsp_goto(self, data: dict):
method = data["method"]
language_id = data["language_id"]
uri = data["uri"]
line = data["line"]
column = data["column"]
self._client_send_request(method, uri, line, column)
# return "{'response':'some kind of response'}"
def _lsp_completion(self, data: dict):
method = data["method"]
language_id = data["language_id"]
uri = data["uri"]
line = data["line"]
column = data["column"]
# self._client_send_request(method, uri, line, column)

3
src/core/widgets/buttons/workspace_folder_choser_button.py
View File

@ -27,7 +27,8 @@ class WorkspaceFolderChoserButton(Gtk.FileChooserButton):
self.set_title("Chose Workspace")
self.set_action( Gtk.FileChooserAction.SELECT_FOLDER )
self.set_uri("file:///home/abaddon/Coding/Projects/Active/Python_Projects/testing/lsp_manager")
# self.set_uri("file:///home/abaddon/Coding/Projects/Active/Python_Projects/testing/lsp_manager")
self.set_uri("file:///home/abaddon/Coding/Projects/Active/Python_Projects/000_Usable/gtk/LSP-Manager")
def _setup_signals(self):

30
src/libs/dto/lsp_message_structs.py
View File

@ -48,17 +48,35 @@ class ClientNotification(object):
self.params = params
@dataclass
class LSPResponse(object):
class LSPResponseRequest(object):
"""
Constructs a new LSP Response instance.
Constructs a new LSP Response Request instance.
:param str method: The type of lsp notification being made.
:param id result: The id of the given message.
:param dict result: The arguments of the given method.
"""
jsonrpc: str
id: int or None
result: {}
id: int
result: dict
@dataclass
class LSPResponseNotification(object):
"""
Constructs a new LSP Response Notification instance.
:param str method: The type of lsp notification being made.
:params dict result: The arguments of the given method.
"""
jsonrpc: str
method: str
params: dict
class MessageTypes(ClientRequest, ClientNotification, LSPResponse):
class MessageTypes(ClientRequest, ClientNotification, LSPResponseRequest, LSPResponseNotification):
...
class ClientMessageTypes(ClientRequest, ClientNotification):
...
class LSPResponseTypes(LSPResponseRequest, LSPResponseNotification):
...

13
src/libs/dto/lsp_messages.py
View File

@ -40,6 +40,19 @@ content_part = {
}
}
didopen_notification = {
"method": "textDocument/didOpen",
"params": {
"textDocument": {
"uri": "file://<path>",
"languageId": "python3",
"version": 1,
"text": ""
}
}
}
definition_query = {
"method": "textDocument/definition",

3
src/libs/ipc_server.py
View File

@ -76,6 +76,9 @@ class IPCServer(Singleton):
if file:
event_system.emit("handle-file-from-ipc", file)
conn.close()
break
if "DIR|" in msg:
file = msg.split("DIR|")[1].strip()
if file:

148
src/libs/lsp_endpoint_server.py Normal file
View File

@ -0,0 +1,148 @@
# Python imports
import os
import threading
import time
import json
import base64
from multiprocessing.connection import Client
from multiprocessing.connection import Listener
# Lib imports
# Application imports
from .singleton import Singleton
class LSPEndpointServer(Singleton):
""" Create a listener so that LSP Clients can communicate to this instances and get responses back. """
def __init__(self, ipc_address: str = '127.0.0.1', conn_type: str = "socket"):
self.is_ipc_alive = False
self._ipc_port = 4848
self._ipc_address = ipc_address
self._conn_type = conn_type
self._ipc_authkey = b'' + bytes(f'lsp-manager-endpoint-ipc', 'utf-8')
self._client_ipc_authkey = b'' + bytes(f'lsp-client-endpoint-ipc', 'utf-8')
self._ipc_timeout = 15.0
if conn_type == "socket":
self._ipc_address = f'/tmp/lsp-manager-endpoint-ipc.sock'
self._client_ipc_address = f'/tmp/lsp-client-endpoint-ipc.sock'
elif conn_type == "full_network":
self._ipc_address = '0.0.0.0'
elif conn_type == "full_network_unsecured":
self._ipc_authkey = None
self._ipc_address = '0.0.0.0'
elif conn_type == "local_network_unsecured":
self._ipc_authkey = None
self._subscribe_to_events()
def _subscribe_to_events(self):
event_system.subscribe("respond-to-client", self.send_client_ipc_message)
def create_ipc_listener(self) -> None:
if self._conn_type == "socket":
if os.path.exists(self._ipc_address) and settings_manager.is_dirty_start():
os.unlink(self._ipc_address)
listener = Listener(address=self._ipc_address, family="AF_UNIX", authkey=self._ipc_authkey)
elif "unsecured" not in self._conn_type:
listener = Listener((self._ipc_address, self._ipc_port), authkey=self._ipc_authkey)
else:
listener = Listener((self._ipc_address, self._ipc_port))
self.is_ipc_alive = True
self._run_ipc_loop(listener)
@daemon_threaded
def _run_ipc_loop(self, listener) -> None:
# NOTE: Not thread safe if using with Gtk. Need to import GLib and use idle_add
while True:
try:
conn = listener.accept()
start_time = time.perf_counter()
self._handle_ipc_message(conn, start_time)
except Exception as e:
logger.debug( repr(e) )
listener.close()
def _handle_ipc_message(self, conn, start_time) -> None:
while True:
msg = conn.recv()
logger.debug(msg)
if "CLIENT|" in msg:
data = msg.split("CLIENT|")[1].strip()
if data:
data_str = base64.b64decode(data.encode("utf-8")).decode("utf-8")
json_blob = json.loads(data_str)
event_system.emit(json_blob["method"], (json_blob,))
conn.close()
break
if msg in ['close connection', 'close server']:
conn.close()
break
# NOTE: Not perfect but insures we don't lock up the connection for too long.
end_time = time.perf_counter()
if (end_time - start_time) > self._ipc_timeout:
conn.close()
break
def send_client_ipc_message(self, message: str = "Empty Data...") -> None:
try:
if self._conn_type == "socket":
conn = Client(address=self._client_ipc_address, family="AF_UNIX", authkey=self._client_ipc_authkey)
elif "unsecured" not in self._conn_type:
conn = Client((self._ipc_address, self._ipc_port), authkey=self._ipc_authkey)
else:
conn = Client((self._ipc_address, self._ipc_port))
conn.send( f"MANAGER|{ base64.b64encode(message.encode("utf-8")).decode("utf-8") }" )
conn.close()
except ConnectionRefusedError as e:
logger.error("Connection refused...")
except Exception as e:
logger.error( repr(e) )
def send_ipc_message(self, message: str = "Empty Data...") -> None:
try:
if self._conn_type == "socket":
conn = Client(address=self._ipc_address, family="AF_UNIX", authkey=self._ipc_authkey)
elif "unsecured" not in self._conn_type:
conn = Client((self._ipc_address, self._ipc_port), authkey=self._ipc_authkey)
else:
conn = Client((self._ipc_address, self._ipc_port))
conn.send(message)
conn.close()
except ConnectionRefusedError as e:
logger.error("Connection refused...")
except Exception as e:
logger.error( repr(e) )
def send_test_ipc_message(self, message: str = "Empty Data...") -> None:
try:
if self._conn_type == "socket":
conn = Client(address=self._ipc_address, family="AF_UNIX", authkey=self._ipc_authkey)
elif "unsecured" not in self._conn_type:
conn = Client((self._ipc_address, self._ipc_port), authkey=self._ipc_authkey)
else:
conn = Client((self._ipc_address, self._ipc_port))
conn.send(message)
conn.close()
except ConnectionRefusedError as e:
if self._conn_type == "socket":
logger.error("LSP Socket no longer valid.... Removing.")
os.unlink(self._ipc_address)
except Exception as e:
logger.error( repr(e) )