Merge pull request #95 from HenningHolmDE/use_lsp-server_crate

Use lsp-server crate for RPC channel handling

Author: kraigher

vhdl_ls/Cargo.toml

@@ -15,14 +15,14 @@ edition = "2018"
 
 [dependencies]
 vhdl_lang = { version = "^0", path = "../vhdl_lang"}
-jsonrpc-core = "^9"
 serde_json = "^1"
 serde = "^1"
 lsp-types = "^0.63"
 fnv = "^1"
 log = "0.4.6"
 env_logger = "0.6.0"
 clap = "^2"
+lsp-server = "^0.3"
 
 [dev-dependencies]
 tempfile = "^3"

vhdl_ls/src/stdio_server.rs

@@ -4,210 +4,194 @@
 //
 // Copyright (c) 2018, Olof Kraigher olof.kraigher@gmail.com
 
-use jsonrpc_core::request::Notification;
-use jsonrpc_core::{IoHandler, Params};
-use std::io::prelude::*;
-use std::io::{self, BufRead};
+//! This module handles setting up `VHDLServer` for `stdio` communication.
+//! It also contains the main event loop for handling incoming messages from the LSP client and
+//! dispatching them to the appropriate server methods.
 
-use std::sync::mpsc::{sync_channel, SyncSender};
-use std::thread::spawn;
-
-use std::sync::{Arc, Mutex};
+use lsp_server::Connection;
+use lsp_types::{
+    notification::{self, Notification},
+    request, InitializeParams,
+};
+use std::rc::Rc;
 
 use crate::rpc_channel::RpcChannel;
 use crate::vhdl_server::VHDLServer;
 
+/// Set up the IO channel for `stdio` and start the VHDL language server.
 pub fn start() {
-    let (request_sender, request_receiver) = sync_channel(1);
-    let (response_sender, response_receiver) = sync_channel(1);
-    let mut io = IoHandler::new();
-
-    // @TODO handle jsonrpc synchronously
-    let lang_server = Arc::new(Mutex::new(VHDLServer::new(response_sender.clone())));
-    let server = lang_server.clone();
-    io.add_method("initialize", move |params: Params| {
-        let value = server.lock().unwrap().initialize_request(params.parse()?)?;
-        serde_json::to_value(value).map_err(|_| jsonrpc_core::Error::internal_error())
-    });
-
-    let server = lang_server.clone();
-    io.add_method("shutdown", move |params: Params| {
-        server.lock().unwrap().shutdown_server(params.parse()?)?;
-        Ok(serde_json::Value::Null)
-    });
-
-    let server = lang_server.clone();
-    io.add_notification("initialized", move |_params: Params| {
-        server.lock().unwrap().initialized_notification()
-    });
-
-    let server = lang_server.clone();
-    io.add_notification("exit", move |_params: Params| {
-        server.lock().unwrap().exit_notification(())
-    });
-
-    let server = lang_server.clone();
-    io.add_notification("textDocument/didChange", move |params: Params| {
-        server
-            .lock()
-            .unwrap()
-            .text_document_did_change_notification(&params.parse().unwrap())
-    });
-
-    let server = lang_server.clone();
-    io.add_notification("textDocument/didOpen", move |params: Params| {
-        server
-            .lock()
-            .unwrap()
-            .text_document_did_open_notification(&params.parse().unwrap())
-    });
-
-    let server = lang_server.clone();
-    io.add_method("textDocument/declaration", move |params: Params| {
-        let value = server
-            .lock()
-            .unwrap()
-            .text_document_declaration(&params.parse().unwrap());
-        serde_json::to_value(value).map_err(|_| jsonrpc_core::Error::internal_error())
-    });
-
-    let server = lang_server.clone();
-    io.add_method("textDocument/definition", move |params: Params| {
-        let value = server
-            .lock()
-            .unwrap()
-            .text_document_definition(&params.parse().unwrap());
-        serde_json::to_value(value).map_err(|_| jsonrpc_core::Error::internal_error())
-    });
-
-    let server = lang_server;
-    io.add_method("textDocument/references", move |params: Params| {
-        let value = server
-            .lock()
-            .unwrap()
-            .text_document_references(&params.parse().unwrap());
-        serde_json::to_value(value).map_err(|_| jsonrpc_core::Error::internal_error())
-    });
-
-    // Spawn thread to read requests from stdin
-    spawn(move || {
-        let stdin = io::stdin();
-        loop {
-            let request = read_request(&mut stdin.lock());
-            match request_sender.send(request) {
-                Ok(_) => continue,
-                Err(_) => {
-                    info!("Channel hung up. Unlocking stdin handle.");
-                    break;
-                }
-            }
-        }
-    });
-
-    // Spawn thread to write notifications to stdout
-    spawn(move || {
-        let mut stdout = io::stdout();
-        loop {
-            match response_receiver.recv() {
-                Ok(response) => {
-                    send_response(&mut stdout, &response);
-                }
-                Err(_) => {
-                    info!("Channel hung up.");
-                    break;
-                }
+    let (connection, io_threads) = Connection::stdio();
+    let connection = Rc::new(connection);
+    let mut server = VHDLServer::new(connection.clone());
+
+    handle_initialization(&connection, &mut server);
+    main_event_loop(connection, server);
+    io_threads.join().unwrap();
+}
+
+/// Wait for initialize request from the client and let the server respond to it.
+fn handle_initialization<T: RpcChannel + Clone>(
+    connection: &Connection,
+    server: &mut VHDLServer<T>,
+) {
+    let (initialize_id, initialize_params) = connection.initialize_start().unwrap();
+    let initialize_params = serde_json::from_value::<InitializeParams>(initialize_params).unwrap();
+    let initialize_result = server.initialize_request(initialize_params);
+    connection
+        .initialize_finish(
+            initialize_id,
+            serde_json::to_value(initialize_result).unwrap(),
+        )
+        .unwrap();
+
+    server.initialized_notification();
+}
+
+/// Main event loop handling incoming messages from the client.
+fn main_event_loop<T: RpcChannel + Clone>(connection: Rc<Connection>, mut server: VHDLServer<T>) {
+    info!("Language server initialized, waiting for messages ...");
+    while let Ok(message) = connection.receiver.recv() {
+        trace!("Received message: {:?}", message);
+        if let lsp_server::Message::Notification(notification) = &message {
+            if notification.method == notification::Exit::METHOD {
+                return;
             }
         }
-    });
-
-    loop {
-        match request_receiver.recv() {
-            Ok(request) => {
-                let response = io.handle_request_sync(&request);
-                if let Some(response) = response {
-                    response_sender.send(response).unwrap();
-                }
+        match message {
+            lsp_server::Message::Request(request) => {
+                handle_request(&mut server, connection.as_ref(), request)
             }
-            Err(_) => {
-                info!("Channel hung up.");
-                break;
+            lsp_server::Message::Notification(notification) => {
+                handle_notification(&mut server, notification);
             }
-        }
+            lsp_server::Message::Response(response) => handle_response(&mut server, response),
+        };
     }
 }
 
-fn read_request(reader: &mut dyn BufRead) -> String {
-    let content_length = read_header(reader);
-
-    let mut request = String::new();
-    reader
-        .take(content_length)
-        .read_to_string(&mut request)
-        .unwrap();
-    trace!("GOT REQUEST: {:?}", request);
-    request
-}
-
-fn send_response(writer: &mut dyn Write, response: &str) {
-    trace!("SEND RESPONSE: {:?}", response);
-    writeln!(writer, "Content-Length: {}\r", response.len()).unwrap();
-    writeln!(writer, "\r").unwrap();
-    write!(writer, "{}", response).unwrap();
-    writer.flush().expect("Could not flush stdout");
+/// Send responses (to requests sent by the client) back to the client.
+fn send_response(connection: &Connection, response: lsp_server::Response) {
+    trace!("Sending response: {:?}", response);
+    connection.sender.send(response.into()).unwrap();
 }
 
-impl RpcChannel for SyncSender<String> {
+impl RpcChannel for Rc<Connection> {
+    /// Send notifications to the client.
     fn send_notification(
         &self,
         method: impl Into<String>,
         notification: impl serde::ser::Serialize,
     ) {
-        let params_json = match serde_json::to_value(notification).unwrap() {
-            serde_json::Value::Object(map) => map,
-            map => panic!("{:?}", map),
-        };
-
-        let notification_json = Notification {
-            jsonrpc: Some(jsonrpc_core::Version::V2),
+        let notification = lsp_server::Notification {
             method: method.into(),
-            params: Params::Map(params_json),
+            params: serde_json::to_value(notification).unwrap(),
         };
 
-        self.send(serde_json::to_string(&notification_json).unwrap())
-            .unwrap();
+        trace!("Sending notification: {:?}", notification);
+        self.sender.send(notification.into()).unwrap();
     }
 }
 
-fn read_header(reader: &mut dyn BufRead) -> u64 {
-    let mut buffer = String::new();
-    reader.read_line(&mut buffer).unwrap();
-    let fields = buffer.trim_end().split(": ").collect::<Vec<&str>>();
-    if fields.get(0) != Some(&"Content-Length") {
-        trace!("{:?}", fields);
-        panic!();
+/// Handle incoming requests from the client.
+fn handle_request<T: RpcChannel + Clone>(
+    server: &mut VHDLServer<T>,
+    connection: &Connection,
+    request: lsp_server::Request,
+) {
+    fn extract<R>(
+        request: lsp_server::Request,
+    ) -> Result<(lsp_server::RequestId, R::Params), lsp_server::Request>
+    where
+        R: request::Request,
+        R::Params: serde::de::DeserializeOwned,
+    {
+        request.extract(R::METHOD)
     }
-    let content_length = fields[1].parse::<u64>().unwrap();
 
-    let mut buffer = String::new();
-    reader.read_line(&mut buffer).unwrap();
-    if buffer == "\r\n" {
-        return content_length;
-    }
+    trace!("Handling request: {:?}", request);
+    let request = match extract::<request::GotoDeclaration>(request) {
+        Ok((id, params)) => {
+            let result = server.text_document_declaration(&params);
+            send_response(connection, lsp_server::Response::new_ok(id, result));
+            return;
+        }
+        Err(request) => request,
+    };
+    let request = match extract::<request::GotoDefinition>(request) {
+        Ok((id, params)) => {
+            let result = server.text_document_definition(&params);
+            send_response(connection, lsp_server::Response::new_ok(id, result));
+            return;
+        }
+        Err(request) => request,
+    };
+    let request = match extract::<request::References>(request) {
+        Ok((id, params)) => {
+            let result = server.text_document_references(&params);
+            send_response(connection, lsp_server::Response::new_ok(id, result));
+            return;
+        }
+        Err(request) => request,
+    };
+    let request = match extract::<request::Shutdown>(request) {
+        Ok((id, _params)) => {
+            server.shutdown_server();
+            send_response(connection, lsp_server::Response::new_ok(id, ()));
+            return;
+        }
+        Err(request) => request,
+    };
+
+    debug!("Unhandled request: {:?}", request);
+    send_response(
+        connection,
+        lsp_server::Response::new_err(
+            request.id,
+            lsp_server::ErrorCode::MethodNotFound as i32,
+            "Unknown request".to_string(),
+        ),
+    );
+}
 
-    let fields = buffer.trim_end().split(": ").collect::<Vec<&str>>();
-    if fields.get(0) != Some(&"Content-Type") {
-        trace!("{:?}", fields);
-        panic!();
-    } else {
-        trace!("got Content-Type: {}", &fields[1]);
+/// Handle incoming notifications from the client.
+fn handle_notification<T: RpcChannel + Clone>(
+    server: &mut VHDLServer<T>,
+    notification: lsp_server::Notification,
+) {
+    fn extract<N>(
+        notification: lsp_server::Notification,
+    ) -> Result<N::Params, lsp_server::Notification>
+    where
+        N: notification::Notification,
+        N::Params: serde::de::DeserializeOwned,
+    {
+        notification.extract(N::METHOD)
     }
 
-    let mut buffer = String::new();
-    reader.read_line(&mut buffer).unwrap();
-    if buffer != "\r\n" {
-        trace!("{:?}", buffer);
-        panic!();
+    trace!("Handling notification: {:?}", notification);
+    let notification = match extract::<notification::DidChangeTextDocument>(notification) {
+        Ok(params) => return server.text_document_did_change_notification(&params),
+        Err(notification) => notification,
+    };
+    let notification = match extract::<notification::DidOpenTextDocument>(notification) {
+        Ok(params) => return server.text_document_did_open_notification(&params),
+        Err(notification) => notification,
+    };
+    let notification = match extract::<notification::Exit>(notification) {
+        Ok(_params) => return server.exit_notification(),
+        Err(notification) => notification,
+    };
+
+    if !notification.method.starts_with("$/") {
+        debug!("Unhandled notification: {:?}", notification);
     }
+}
 
-    content_length
+/// Handle incoming responses (to requests sent by us) from the client.
+fn handle_response<T: RpcChannel + Clone>(
+    _server: &mut VHDLServer<T>,
+    response: lsp_server::Response,
+) {
+    trace!("Handling response: {:?}", response);
+    debug!("Unhandled response: {:?}", response);
 }