"""Erlang Language Server implementation using Erlang LS.""" import logging import os import pathlib import shutil import subprocess import threading import time from overrides import override from solidlsp.ls import SolidLanguageServer from solidlsp.ls_config import LanguageServerConfig from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo from solidlsp.settings import SolidLSPSettings log = logging.getLogger(__name__) class ErlangLanguageServer(SolidLanguageServer): """Language server for Erlang using Erlang LS.""" def __init__(self, config: LanguageServerConfig, repository_root_path: str, solidlsp_settings: SolidLSPSettings): """ Creates an ErlangLanguageServer instance. This class is not meant to be instantiated directly. Use LanguageServer.create() instead. """ self.erlang_ls_path = shutil.which("erlang_ls") if not self.erlang_ls_path: raise RuntimeError("Erlang LS not found. Install from: https://github.com/erlang-ls/erlang_ls") if not self._check_erlang_installation(): raise RuntimeError("Erlang/OTP not found. Install from: https://www.erlang.org/downloads") super().__init__( config, repository_root_path, ProcessLaunchInfo(cmd=[self.erlang_ls_path, "--transport", "stdio"], cwd=repository_root_path), "erlang", solidlsp_settings, ) # Add server readiness tracking like Elixir self.server_ready = threading.Event() # Set generous timeout for Erlang LS initialization self.set_request_timeout(120.0) def _check_erlang_installation(self) -> bool: """Check if Erlang/OTP is available.""" try: result = subprocess.run(["erl", "-version"], check=False, capture_output=True, text=True, timeout=10) return result.returncode == 0 except (subprocess.SubprocessError, FileNotFoundError): return False @classmethod def _get_erlang_version(cls) -> str | None: """Get the installed Erlang/OTP version or None if not found.""" try: result = subprocess.run(["erl", "-version"], check=False, capture_output=True, text=True, timeout=10) if result.returncode == 0: return result.stderr.strip() # erl -version outputs to stderr except (subprocess.SubprocessError, FileNotFoundError): return None return None @classmethod def _check_rebar3_available(cls) -> bool: """Check if rebar3 build tool is available.""" try: result = subprocess.run(["rebar3", "version"], check=False, capture_output=True, text=True, timeout=10) return result.returncode == 0 except (subprocess.SubprocessError, FileNotFoundError): return False def _start_server(self) -> None: """Start Erlang LS server process with proper initialization waiting.""" def register_capability_handler(params: dict) -> None: return def window_log_message(msg: dict) -> None: """Handle window/logMessage notifications from Erlang LS""" message_text = msg.get("message", "") log.info(f"LSP: window/logMessage: {message_text}") # Look for Erlang LS readiness signals # Common patterns: "Started Erlang LS", "initialized", "ready" readiness_signals = [ "Started Erlang LS", "server started", "initialized", "ready to serve requests", "compilation finished", "indexing complete", ] message_lower = message_text.lower() for signal in readiness_signals: if signal.lower() in message_lower: log.info(f"Erlang LS readiness signal detected: {message_text}") self.server_ready.set() break def do_nothing(params: dict) -> None: return def check_server_ready(params: dict) -> None: """Handle $/progress notifications from Erlang LS as fallback.""" value = params.get("value", {}) # Check for initialization completion progress if value.get("kind") == "end": message = value.get("message", "") if any(word in message.lower() for word in ["initialized", "ready", "complete"]): log.info("Erlang LS initialization progress completed") # Set as fallback if no window/logMessage was received if not self.server_ready.is_set(): self.server_ready.set() # Set up notification handlers self.server.on_request("client/registerCapability", register_capability_handler) self.server.on_notification("window/logMessage", window_log_message) self.server.on_notification("$/progress", check_server_ready) self.server.on_notification("window/workDoneProgress/create", do_nothing) self.server.on_notification("$/workDoneProgress", do_nothing) self.server.on_notification("textDocument/publishDiagnostics", do_nothing) log.info("Starting Erlang LS server process") self.server.start() # Send initialize request initialize_params = { "processId": None, "rootPath": self.repository_root_path, "rootUri": pathlib.Path(self.repository_root_path).as_uri(), "capabilities": { "textDocument": { "synchronization": {"didSave": True}, "completion": {"dynamicRegistration": True}, "definition": {"dynamicRegistration": True}, "references": {"dynamicRegistration": True}, "documentSymbol": {"dynamicRegistration": True}, "hover": {"dynamicRegistration": True}, } }, } log.info("Sending initialize request to Erlang LS") init_response = self.server.send.initialize(initialize_params) # type: ignore[arg-type] # Verify server capabilities if "capabilities" in init_response: log.info(f"Erlang LS capabilities: {list(init_response['capabilities'].keys())}") self.server.notify.initialized({}) # Wait for Erlang LS to be ready - adjust timeout based on environment is_ci = os.getenv("CI") == "true" or os.getenv("GITHUB_ACTIONS") == "true" is_macos = os.uname().sysname == "Darwin" if hasattr(os, "uname") else False # macOS in CI can be particularly slow for language server startup if is_ci and is_macos: ready_timeout = 240.0 # 4 minutes for macOS CI env_desc = "macOS CI" elif is_ci: ready_timeout = 180.0 # 3 minutes for other CI env_desc = "CI" else: ready_timeout = 60.0 # 1 minute for local env_desc = "local" log.info(f"Waiting up to {ready_timeout} seconds for Erlang LS readiness ({env_desc} environment)...") if self.server_ready.wait(timeout=ready_timeout): log.info("Erlang LS is ready and available for requests") # Add settling period for indexing - adjust based on environment settling_time = 15.0 if is_ci else 5.0 log.info(f"Allowing {settling_time} seconds for Erlang LS indexing to complete...") time.sleep(settling_time) log.info("Erlang LS settling period complete") else: # Set ready anyway and continue - Erlang LS might not send explicit ready messages log.warning(f"Erlang LS readiness timeout reached after {ready_timeout}s, proceeding anyway (common in CI)") self.server_ready.set() # Still give some time for basic initialization even without explicit readiness signal basic_settling_time = 20.0 if is_ci else 10.0 log.info(f"Allowing {basic_settling_time} seconds for basic Erlang LS initialization...") time.sleep(basic_settling_time) log.info("Basic Erlang LS initialization period complete") @override def is_ignored_dirname(self, dirname: str) -> bool: # For Erlang projects, we should ignore: # - _build: rebar3 build artifacts # - deps: dependencies # - ebin: compiled beam files # - .rebar3: rebar3 cache # - logs: log files # - node_modules: if the project has JavaScript components return super().is_ignored_dirname(dirname) or dirname in [ "_build", "deps", "ebin", ".rebar3", "logs", "node_modules", "_checkouts", "cover", ] def is_ignored_filename(self, filename: str) -> bool: """Check if a filename should be ignored.""" # Ignore compiled BEAM files if filename.endswith(".beam"): return True # Don't ignore Erlang source files, header files, or configuration files return False