Refatored lsp controller file layout

src/core/controllers/lsp/__init__.py

@@ -0,0 +1,3 @@
+"""
+    LSP Controller Module
+"""

src/core/controllers/lsp/lsp_controller.py

@@ -0,0 +1,91 @@
+# Python imports
+import threading
+
+# Lib imports
+
+# Application imports
+from libs.dto.lsp_messages import get_message_str
+from libs.dto.lsp_message_structs import LSPResponseTypes, ClientRequest, ClientNotification
+from .lsp_controller_stdin_stdout import LSPControllerSTDInSTDOut
+
+
+
+def _log_list():
+    def add_log_entry(title: str, message: any):
+        ...
+
+    def clear():
+        ...
+
+
+
+class LSPController(LSPControllerSTDInSTDOut):
+    def __init__(self):
+        super(LSPController).__init__()
+
+        self._language      = "python3"
+
+        # https://github.com/microsoft/multilspy/tree/main/src/multilspy/language_servers
+        # initialize-params-slim.json was created off of jedi_language_server one
+        self._init_params   = settings_manager.get_lsp_init_data()
+
+        self._start_command = None
+        self._lsp_pid       = -1
+        self._message_id    = 0
+
+        self.read_lock      = threading.Lock()
+        self.write_lock     = threading.Lock()
+        self.log_list       = _log_list
+        self.request_list   = {}
+
+        self._subscribe_to_events()
+
+
+    def _subscribe_to_events(self):
+        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_definition)
+        event_system.subscribe("textDocument/completion", self._lsp_completion)
+
+
+    def set_language(self, language):
+        self._language = language
+
+    def set_log_list(self, log_list):
+        self.log_list = log_list
+
+    def set_start_command(self, start_command: list):
+        self._start_command = start_command
+
+    def unset_start_command(self):
+        self._start_command = None
+
+    def send_notification(self, method: str, params: {} = {}):
+        self._send_message( ClientNotification(method, params) )
+
+    def send_request(self, method: str, params: {} = {}):
+        self._send_message( ClientRequest(self._message_id, method, params) )
+
+        self.request_list[self._message_id] = {"method": method}
+        self._message_id += 1
+
+    def get_message_id(self) -> int:
+        return self._message_id
+
+    def start_stop_lsp(self):
+        if self._lsp_pid == -1:
+            pid = self.start_lsp()
+            if not pid: return
+
+            self._lsp_pid = pid
+            self._monitor_lsp_response()
+        else:
+            self.stop_lsp()
+            self.log_list.clear()
+
+    def handle_lsp_response(self, lsp_response: LSPResponseTypes):
+        self._monitor_lsp_response()
+        self.log_list.add_log_entry("LSP Response", lsp_response)
+        event_system.emit("respond-to-client", (get_message_str(lsp_response),))

src/core/controllers/lsp/lsp_controller_base.py

@@ -0,0 +1,19 @@
+# Python imports
+
+# Lib imports
+
+# Application imports
+from .lsp_controller_events import LSPControllerEvents
+from libs.dto.lsp_message_structs import ClientRequest, ClientNotification
+
+
+
+class LSPControllerBase(LSPControllerEvents):
+    def _send_message(self, data: ClientRequest or ClientNotification):
+        raise NotImplementedError
+
+    def start_lsp(self):
+        raise NotImplementedError
+
+    def stop_lsp(self):
+        raise NotImplementedError

src/core/controllers/lsp/lsp_controller_events.py

@@ -0,0 +1,108 @@
+# Python imports
+import os
+
+# Lib imports
+from gi.repository import GLib
+
+# Application imports
+from libs.dto.lsp_messages import get_message_obj
+from libs.dto.lsp_messages import didopen_notification
+from libs.dto.lsp_messages import didsave_notification
+from libs.dto.lsp_messages import didclose_notification
+from libs.dto.lsp_messages import didchange_notification
+from libs.dto.lsp_messages import completion_request
+from libs.dto.lsp_messages import definition_request
+from libs.dto.lsp_messages import references_request
+from libs.dto.lsp_messages import symbols_request
+
+
+
+class LSPControllerEvents:
+    def send_initialize_message(self, init_ops: str, workspace_file: str, workspace_uri: str):
+        folder_name = os.path.basename(workspace_file)
+
+        self._init_params["processId"]        = settings_manager.get_app_pid()
+        self._init_params["rootPath"]         = workspace_file
+        self._init_params["rootUri"]          = workspace_uri
+        self._init_params["workspaceFolders"] = [
+            {
+                "name": folder_name,
+                "uri": workspace_uri
+            }
+        ]
+
+        self._init_params["initializationOptions"] = get_message_obj(init_ops)
+        self.send_request("initialize", self._init_params)
+
+    def send_initialized_message(self):
+        self.send_notification("initialized")
+
+    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):
+        method = data["method"]
+        params = didsave_notification["params"]
+
+        params["textDocument"]["uri"] = data["uri"]
+        params["text"]                = data["text"]
+
+        GLib.idle_add( self.send_notification, method, params )
+
+    def _lsp_did_close(self, data: dict):
+        method = data["method"]
+        params = didclose_notification["params"]
+
+        params["textDocument"]["uri"] = data["uri"]
+
+        GLib.idle_add( self.send_notification, method, params )
+
+    def _lsp_did_change(self, data: dict):
+        method = data["method"]
+        params = didchange_notification["params"]
+
+        params["textDocument"]["uri"]        = data["uri"]
+        params["textDocument"]["languageId"] = data["language_id"]
+        params["textDocument"]["version"]    = data["version"]
+
+        contentChanges         = params["contentChanges"][0]
+        start                  = contentChanges["range"]["start"]
+        end                    = contentChanges["range"]["end"]
+        contentChanges["text"] = data["text"]
+        start["line"]          = data["line"]
+        start["character"]     = 0
+        end["line"]            = data["line"]
+        end["character"]       = data["column"]
+
+        GLib.idle_add( self.send_notification, method, params )
+
+    def _lsp_definition(self, data: dict):
+        method = data["method"]
+        params = definition_request["params"]
+
+        params["textDocument"]["uri"]        = data["uri"]
+        params["textDocument"]["languageId"] = data["language_id"]
+        params["textDocument"]["version"]    = data["version"]
+        params["position"]["line"]           = data["line"]
+        params["position"]["character"]      = data["column"]
+
+        GLib.idle_add( self.send_request, method, params )
+
+    def _lsp_completion(self, data: dict):
+        method = data["method"]
+        params = completion_request["params"]
+
+        params["textDocument"]["uri"]        = data["uri"]
+        params["textDocument"]["languageId"] = data["language_id"]
+        params["textDocument"]["version"]    = data["version"]
+        params["position"]["line"]           = data["line"]
+        params["position"]["character"]      = data["column"]
+
+        GLib.idle_add( self.send_request, method, params )

src/core/controllers/lsp/lsp_controller_stdin_stdout.py

@@ -0,0 +1,116 @@
+# Python imports
+import subprocess
+
+# 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
+from .lsp_controller_base import LSPControllerBase
+from libs.dto.lsp_message_structs import \
+    LSPResponseTypes, ClientRequest, ClientNotification, LSPResponseRequest, LSPResponseNotification
+
+
+
+
+class LSPControllerSTDInSTDOut(LSPControllerBase):
+    def _send_message(self, data: ClientRequest or ClientNotification):
+        if not data or not hasattr(self, "lsp_process"): return
+
+        message_str  = get_message_str(data)
+        message_size = len(message_str)
+        message      = f"Content-Length: {message_size}\r\n\r\n{message_str}"
+
+        self.log_list.add_log_entry("Client", data)
+
+        with self.write_lock:
+            self.lsp_process.stdin.write( message.encode("utf-8") )
+            self.lsp_process.stdin.flush()
+
+    def start_lsp(self):
+        if not self._start_command: return
+        try:
+            self.lsp_process = subprocess.Popen(
+                self._start_command,
+                stdout = subprocess.PIPE,
+                stdin  = subprocess.PIPE
+            )
+        except Exception as e:
+            self.log_list.add_log_entry(
+                "LSP Client Error",
+                LSPResponseRequest(
+                    "2.0",
+                    None,
+                    {
+                        "error": repr(e)
+                    }
+                )
+            )
+
+            return
+
+        return self.lsp_process.pid
+
+    def stop_lsp(self):
+        if self._lsp_pid == -1: return
+
+        self._lsp_pid    = -1
+        self._message_id = 0
+        self.lsp_process.terminate()
+
+    # https://github.com/sr-lab/coqpyt/blob/master/coqpyt/lsp/json_rpc_endpoint.py#L65
+    # Virtually this whole method unabashedly taken from ^ ...
+    @daemon_threaded
+    def _monitor_lsp_response(self):
+        if not hasattr(self, "lsp_process"): return
+
+        with self.read_lock:
+            message_size = None
+            while True:
+                line = self.lsp_process.stdout.readline()
+                if not line: return None  # Quit listener...
+
+                line = line.decode("utf-8")
+                if not line.endswith("\r\n"):
+                    raise Exception(
+                        "Bad header: missing newline"
+                    )
+                line = line[:-2]  # Strip the "\r\n"
+                if line == "":    # We're done with the headers...
+                    break
+                elif line.startswith(LEN_HEADER):
+                    line = line[len(LEN_HEADER) :]
+                    if not line.isdigit():
+                        raise Exception(
+                            "Bad header: size is not int",
+                        )
+                    message_size = int(line)
+                elif line.startswith(TYPE_HEADER):
+                    # Not doing anything with type header, currently...
+                    pass
+                else:
+                    line = line.split(LEN_HEADER)
+                    if len(line) == 2:
+                        message_size = line[1]
+                    raise Exception(
+                        "Bad header: unknown header"
+                    )
+
+            if not message_size: return
+
+            _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)