import json
import logging
import os
import pathlib
import subprocess
import threading
from overrides import override
from solidlsp.ls import SolidLanguageServer
from solidlsp.ls_config import LanguageServerConfig
from solidlsp.ls_logger import LanguageServerLogger
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
class Gopls(SolidLanguageServer):
"""
Provides Go specific instantiation of the LanguageServer class using gopls.
"""
@override
def is_ignored_dirname(self, dirname: str) -> bool:
# For Go projects, we should ignore:
# - vendor: third-party dependencies vendored into the project
# - node_modules: if the project has JavaScript components
# - dist/build: common output directories
return super().is_ignored_dirname(dirname) or dirname in ["vendor", "node_modules", "dist", "build"]
@staticmethod
def _get_go_version():
"""Get the installed Go version or None if not found."""
try:
result = subprocess.run(["go", "version"], capture_output=True, text=True, check=False)
if result.returncode == 0:
return result.stdout.strip()
except FileNotFoundError:
return None
return None
@staticmethod
def _get_gopls_version():
"""Get the installed gopls version or None if not found."""
try:
result = subprocess.run(["gopls", "version"], capture_output=True, text=True, check=False)
if result.returncode == 0:
return result.stdout.strip()
except FileNotFoundError:
return None
return None
@classmethod
def setup_runtime_dependency(cls):
"""
Check if required Go runtime dependencies are available.
Raises RuntimeError with helpful message if dependencies are missing.
"""
go_version = cls._get_go_version()
if not go_version:
raise RuntimeError(
"Go is not installed. Please install Go from https://golang.org/doc/install and make sure it is added to your PATH."
)
gopls_version = cls._get_gopls_version()
if not gopls_version:
raise RuntimeError(
"Found a Go version but gopls is not installed.\n"
"Please install gopls as described in https://pkg.go.dev/golang.org/x/tools/gopls#section-readme\n\n"
"After installation, make sure it is added to your PATH (it might be installed in a different location than Go)."
)
return True
def __init__(self, config: LanguageServerConfig, logger: LanguageServerLogger, repository_root_path: str):
self.setup_runtime_dependency()
super().__init__(
config,
logger,
repository_root_path,
ProcessLaunchInfo(cmd="gopls", cwd=repository_root_path),
"go",
)
self.server_ready = threading.Event()
self.request_id = 0
def _get_initialize_params(self, repository_absolute_path: str) -> InitializeParams:
"""
Returns the initialize params for the TypeScript Language Server.
"""
with open(os.path.join(os.path.dirname(__file__), "initialize_params.json"), encoding="utf-8") as f:
d = json.load(f)
del d["_description"]
d["processId"] = os.getpid()
assert d["rootPath"] == "$rootPath"
d["rootPath"] = repository_absolute_path
assert d["rootUri"] == "$rootUri"
d["rootUri"] = pathlib.Path(repository_absolute_path).as_uri()
assert d["workspaceFolders"][0]["uri"] == "$uri"
d["workspaceFolders"][0]["uri"] = pathlib.Path(repository_absolute_path).as_uri()
assert d["workspaceFolders"][0]["name"] == "$name"
d["workspaceFolders"][0]["name"] = os.path.basename(repository_absolute_path)
return d
def _start_server(self):
"""Start gopls server process"""
def register_capability_handler(params):
return
def window_log_message(msg):
self.logger.log(f"LSP: window/logMessage: {msg}", logging.INFO)
def do_nothing(params):
return
self.server.on_request("client/registerCapability", register_capability_handler)
self.server.on_notification("window/logMessage", window_log_message)
self.server.on_notification("$/progress", do_nothing)
self.server.on_notification("textDocument/publishDiagnostics", do_nothing)
self.logger.log("Starting gopls server process", logging.INFO)
self.server.start()
initialize_params = self._get_initialize_params(self.repository_root_path)
self.logger.log(
"Sending initialize request from LSP client to LSP server and awaiting response",
logging.INFO,
)
init_response = self.server.send.initialize(initialize_params)
# Verify server capabilities
assert "textDocumentSync" in init_response["capabilities"]
assert "completionProvider" in init_response["capabilities"]
assert "definitionProvider" in init_response["capabilities"]
self.server.notify.initialized({})
self.completions_available.set()
# gopls server is typically ready immediately after initialization
self.server_ready.set()
self.server_ready.wait()
