使用python调用gopls对Go语言进行分析

package main

import "fmt"

func greet(name string) string {
	return "Hello, " + name + "!"
}

func main() {
	message := greet("World")
	fmt.Println(message)
}

import json
import subprocess
import os
import logging
import time
import threading
import queue
 
logging.basicConfig(filename='lsp_client.log', level=logging.DEBUG, 
                    format='%(asctime)s - %(levelname)s - %(message)s')
 
class LSPClient:
    def __init__(self, compile_commands_dir=None, extra_args=None):
        self.lsp_process = None
        self.request_queue = queue.Queue()
        self.response_queue = queue.Queue()
        self.next_id = 1
        self.running = False
        self.pending_requests = {}
        self.compile_commands_dir = compile_commands_dir
        self.extra_args = extra_args or []
 
 
    def start(self):
        logging.info("Starting LSP server...")
        # 修改 cmd 列表中的 LSP 服务器命令
        cmd = ["gopls", "serve"] 
 
        if self.compile_commands_dir:
            cmd.append(f"--compile-commands-dir={self.compile_commands_dir}")
        cmd.extend(self.extra_args)
        
        self.lsp_process = subprocess.Popen(
            cmd,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            bufsize=1,
            cwd=os.getcwd()
        )
        logging.info(f"LSP server started with PID: {self.lsp_process.pid}")
        
        self.running = True
        threading.Thread(target=self._read_stdout, daemon=True).start()
        threading.Thread(target=self._read_stderr, daemon=True).start()
        threading.Thread(target=self._write_stdin, daemon=True).start()
 
    def stop(self):
        logging.info("Stopping LSP server...")
        self.running = False
        if self.lsp_process:
            self._send_shutdown_request()
            self.lsp_process.terminate()
            self.lsp_process.wait(timeout=5)
        logging.info("LSP server stopped.")
 
    def _send_shutdown_request(self):
        shutdown_request = {
            "jsonrpc": "2.0",
            "id": self._get_next_id(),
            "method": "shutdown"
        }
        self._send_request(shutdown_request)
        # Wait for shutdown response
        time.sleep(1)
        exit_notification = {
            "jsonrpc": "2.0",
            "method": "exit"
        }
        self._send_request(exit_notification)
 
    def _read_stdout(self):
        while self.running:
            try:
                headers = self._read_headers(self.lsp_process.stdout)
                if 'Content-Length' in headers:
                    content_length = int(headers['Content-Length'])
                    content = self.lsp_process.stdout.read(content_length)
                    self._handle_response(content)
            except Exception as e:
                logging.error(f"Error reading stdout: {str(e)}")
 
    def _read_stderr(self):
        for line in self.lsp_process.stderr:
            logging.debug(f"LSP server log: {line.strip()}")
 
    def _write_stdin(self):
        while self.running:
            try:
                request = self.request_queue.get(timeout=0.1)
                self._send_request(request)
            except queue.Empty:
                continue
            except Exception as e:
                logging.error(f"Error writing to stdin: {str(e)}")
 
    def _read_headers(self, stream):
        headers = {}
        while True:
            line = stream.readline().strip()
            if not line:
                break
            key, value = line.split(': ')
            headers[key] = value
        return headers
 
    def _send_request(self, request):
        try:
            content = json.dumps(request)
            headers = f"Content-Length: {len(content)}\r\n\r\n"
            full_message = headers + content
            logging.debug(f"Sending message: {full_message}")
            self.lsp_process.stdin.write(full_message)
            self.lsp_process.stdin.flush()
        except BrokenPipeError:
            logging.error("LSP server process is not available for writing.")
 
    def _handle_response(self, content):
        logging.debug(f"Received response: {content}")
        response = json.loads(content)
        if 'id' in response:
            request_id = response['id']
            if request_id in self.pending_requests:
                self.pending_requests[request_id].put(response)
            else:
                logging.warning(f"Received response for unknown request id: {request_id}")
        elif 'method' in response:
            # Handle server notifications
            if response['method'] == 'textDocument/publishDiagnostics':
                logging.info(f"Diagnostics: {response['params']}")
 
    def _send_request_and_wait(self, request, timeout=5):
        request_id = request['id']
        response_queue = queue.Queue()
        self.pending_requests[request_id] = response_queue
        self.request_queue.put(request)
        try:
            return response_queue.get(timeout=timeout)
        except queue.Empty:
            logging.error(f"Timeout waiting for response to request {request_id}")
            return None
        finally:
            del self.pending_requests[request_id]
 
    def initialize(self):
        request = {
            "jsonrpc": "2.0",
            "id": self._get_next_id(),
            "method": "initialize",
            "params": {
                "processId": os.getpid(),
                "rootUri": f"file://{os.getcwd()}",
                "capabilities": {}
            }
        }
        return self._send_request_and_wait(request)
 
    def initialized(self):
        notification = {
            "jsonrpc": "2.0",
            "method": "initialized",
            "params": {}
        }
        self.request_queue.put(notification)
 
    def did_open(self, file_path):
        with open(file_path, 'r') as file:
            file_content = file.read()
 
        notification = {
            "jsonrpc": "2.0",
            "method": "textDocument/didOpen",
            "params": {
                "textDocument": {
                    "uri": f"file://{file_path}",
                    "languageId": "go",
                    "version": 1,
                    "text": file_content
                }
            }
        }
        self.request_queue.put(notification)
 
    def document_symbols(self, file_path):
        request = {
            "jsonrpc": "2.0",
            "id": self._get_next_id(),
            "method": "textDocument/documentSymbol",
            "params": {
                "textDocument": {"uri": f"file://{file_path}"}
            }
        }
        return self._send_request_and_wait(request)
 
    def definition(self, file_path, line, character):
        request = {
            "jsonrpc": "2.0",
            "id": self._get_next_id(),
            "method": "textDocument/definition",
            "params": {
                "textDocument": {"uri": f"file://{file_path}"},
                "position": {"line": line, "character": character}
            }
        }
        return self._send_request_and_wait(request)
 
    def references(self, file_path, line, character):
        request = {
            "jsonrpc": "2.0",
            "id": self._get_next_id(),
            "method": "textDocument/references",
            "params": {
                "textDocument": {"uri": f"file://{file_path}"},
                "position": {"line": line, "character": character},
                "context": {"includeDeclaration": True}
            }
        }
        return self._send_request_and_wait(request)
 
    def _get_next_id(self):
        id = self.next_id
        self.next_id += 1
        return id
 
def main():
    client = LSPClient()
    try:
        client.start()
        
        init_response = client.initialize()
        if not init_response:
            raise Exception("Failed to initialize LSP server")
        logging.info(f"Initialize response: {init_response}")
        
        client.initialized()
        
        file_path = os.path.join(os.getcwd(), 'example.go')
        client.did_open(file_path)
        
        time.sleep(2)  # Give some time for the server to process the file
        
        symbols_response = client.document_symbols(file_path)
        logging.info(f"Document Symbols response: {symbols_response}")
        
        definition_response = client.definition(file_path, 7, 10)  # 行: 7, 列: 10
        logging.info(f"Definition response: {definition_response}")
        
        references_response = client.references(file_path, 4, 6)  # 行: 4, 列: 6
 
        logging.info(f"References response: {references_response}")
        
    except Exception as e:
        logging.error(f"An error occurred: {str(e)}")
    finally:
        client.stop()
 
if __name__ == "__main__":
    main()

{"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"processId": 33482, "rootUri": "file:///Users/aower/python/lsp_client/go", "capabilities": {}}}