aboutsummaryrefslogtreecommitdiff
path: root/helix-lsp/src
diff options
context:
space:
mode:
Diffstat (limited to 'helix-lsp/src')
-rw-r--r--helix-lsp/src/client.rs61
-rw-r--r--helix-lsp/src/lib.rs27
-rw-r--r--helix-lsp/src/transport.rs180
3 files changed, 117 insertions, 151 deletions
diff --git a/helix-lsp/src/client.rs b/helix-lsp/src/client.rs
index 41c38fd9..9c85beef 100644
--- a/helix-lsp/src/client.rs
+++ b/helix-lsp/src/client.rs
@@ -4,30 +4,22 @@ use crate::{
};
use helix_core::{find_root, ChangeSet, Rope};
-
-// use std::collections::HashMap;
-use std::future::Future;
-use std::sync::atomic::{AtomicU64, Ordering};
-
use jsonrpc_core as jsonrpc;
use lsp_types as lsp;
use serde_json::Value;
-
+use std::future::Future;
use std::process::Stdio;
+use std::sync::atomic::{AtomicU64, Ordering};
use tokio::{
io::{BufReader, BufWriter},
- // prelude::*,
process::{Child, Command},
sync::mpsc::{channel, UnboundedReceiver, UnboundedSender},
};
pub struct Client {
_process: Child,
-
- outgoing: UnboundedSender<Payload>,
- // pub incoming: Receiver<Call>,
- pub request_counter: AtomicU64,
-
+ server_tx: UnboundedSender<Payload>,
+ request_counter: AtomicU64,
capabilities: Option<lsp::ServerCapabilities>,
offset_encoding: OffsetEncoding,
}
@@ -43,40 +35,27 @@ impl Client {
.kill_on_drop(true)
.spawn();
- // use std::io::ErrorKind;
- let mut process = match process {
- Ok(process) => process,
- Err(err) => match err.kind() {
- // ErrorKind::NotFound | ErrorKind::PermissionDenied => {
- // return Err(Error::Other(err.into()))
- // }
- _kind => return Err(Error::Other(err.into())),
- },
- };
+ let mut process = process?;
// TODO: do we need bufreader/writer here? or do we use async wrappers on unblock?
let writer = BufWriter::new(process.stdin.take().expect("Failed to open stdin"));
let reader = BufReader::new(process.stdout.take().expect("Failed to open stdout"));
let stderr = BufReader::new(process.stderr.take().expect("Failed to open stderr"));
- let (incoming, outgoing) = Transport::start(reader, writer, stderr);
+ let (server_rx, server_tx) = Transport::start(reader, writer, stderr);
let client = Self {
_process: process,
-
- outgoing,
- // incoming,
+ server_tx,
request_counter: AtomicU64::new(0),
-
capabilities: None,
- // diagnostics: HashMap::new(),
offset_encoding: OffsetEncoding::Utf8,
};
// TODO: async client.initialize()
// maybe use an arc<atomic> flag
- Ok((client, incoming))
+ Ok((client, server_rx))
}
fn next_request_id(&self) -> jsonrpc::Id {
@@ -106,7 +85,7 @@ impl Client {
}
/// Execute a RPC request on the language server.
- pub async fn request<R: lsp::request::Request>(&self, params: R::Params) -> Result<R::Result>
+ async fn request<R: lsp::request::Request>(&self, params: R::Params) -> Result<R::Result>
where
R::Params: serde::Serialize,
R::Result: core::fmt::Debug, // TODO: temporary
@@ -118,17 +97,20 @@ impl Client {
}
/// Execute a RPC request on the language server.
- pub fn call<R: lsp::request::Request>(
+ fn call<R: lsp::request::Request>(
&self,
params: R::Params,
) -> impl Future<Output = Result<Value>>
where
R::Params: serde::Serialize,
{
- let outgoing = self.outgoing.clone();
+ let server_tx = self.server_tx.clone();
let id = self.next_request_id();
async move {
+ use std::time::Duration;
+ use tokio::time::timeout;
+
let params = serde_json::to_value(params)?;
let request = jsonrpc::MethodCall {
@@ -140,16 +122,13 @@ impl Client {
let (tx, mut rx) = channel::<Result<Value>>(1);
- outgoing
+ server_tx
.send(Payload::Request {
chan: tx,
value: request,
})
.map_err(|e| Error::Other(e.into()))?;
- use std::time::Duration;
- use tokio::time::timeout;
-
timeout(Duration::from_secs(2), rx.recv())
.await
.map_err(|_| Error::Timeout)? // return Timeout
@@ -158,14 +137,14 @@ impl Client {
}
/// Send a RPC notification to the language server.
- pub fn notify<R: lsp::notification::Notification>(
+ fn notify<R: lsp::notification::Notification>(
&self,
params: R::Params,
) -> impl Future<Output = Result<()>>
where
R::Params: serde::Serialize,
{
- let outgoing = self.outgoing.clone();
+ let server_tx = self.server_tx.clone();
async move {
let params = serde_json::to_value(params)?;
@@ -176,7 +155,7 @@ impl Client {
params: Self::value_into_params(params),
};
- outgoing
+ server_tx
.send(Payload::Notification(notification))
.map_err(|e| Error::Other(e.into()))?;
@@ -205,7 +184,7 @@ impl Client {
}),
};
- self.outgoing
+ self.server_tx
.send(Payload::Response(output))
.map_err(|e| Error::Other(e.into()))?;
@@ -216,7 +195,7 @@ impl Client {
// General messages
// -------------------------------------------------------------------------------------------
- pub async fn initialize(&mut self) -> Result<()> {
+ pub(crate) async fn initialize(&mut self) -> Result<()> {
// TODO: delay any requests that are triggered prior to initialize
let root = find_root(None).and_then(|root| lsp::Url::from_file_path(root).ok());
diff --git a/helix-lsp/src/lib.rs b/helix-lsp/src/lib.rs
index af8c9005..5c482774 100644
--- a/helix-lsp/src/lib.rs
+++ b/helix-lsp/src/lib.rs
@@ -1,28 +1,27 @@
mod client;
mod transport;
-pub use jsonrpc_core as jsonrpc;
-pub use lsp_types as lsp;
-
pub use client::Client;
+pub use futures_executor::block_on;
+pub use jsonrpc::Call;
+pub use jsonrpc_core as jsonrpc;
pub use lsp::{Position, Url};
+pub use lsp_types as lsp;
-pub type Result<T> = core::result::Result<T, Error>;
-
+use futures_util::stream::select_all::SelectAll;
use helix_core::syntax::LanguageConfiguration;
-use thiserror::Error;
-
use std::{
collections::{hash_map::Entry, HashMap},
sync::Arc,
};
use serde::{Deserialize, Serialize};
-
+use thiserror::Error;
use tokio_stream::wrappers::UnboundedReceiverStream;
-pub use futures_executor::block_on;
+pub type Result<T> = core::result::Result<T, Error>;
+type LanguageId = String;
#[derive(Error, Debug)]
pub enum Error {
@@ -30,6 +29,8 @@ pub enum Error {
Rpc(#[from] jsonrpc::Error),
#[error("failed to parse: {0}")]
Parse(#[from] serde_json::Error),
+ #[error("IO Error: {0}")]
+ IO(#[from] std::io::Error),
#[error("request timed out")]
Timeout,
#[error("server closed the stream")]
@@ -126,8 +127,6 @@ pub mod util {
}),
)
}
-
- // apply_insert_replace_edit
}
#[derive(Debug, PartialEq, Clone)]
@@ -173,12 +172,6 @@ impl Notification {
}
}
-pub use jsonrpc::Call;
-
-type LanguageId = String;
-
-use futures_util::stream::select_all::SelectAll;
-
pub struct Registry {
inner: HashMap<LanguageId, Arc<Client>>,
diff --git a/helix-lsp/src/transport.rs b/helix-lsp/src/transport.rs
index dd42f7bb..c69c1ccf 100644
--- a/helix-lsp/src/transport.rs
+++ b/helix-lsp/src/transport.rs
@@ -1,15 +1,9 @@
-use std::collections::HashMap;
-use std::io;
-
-use log::{error, info};
-
-use crate::Error;
-
-type Result<T> = core::result::Result<T, Error>;
-
+use crate::Result;
use jsonrpc_core as jsonrpc;
+use log::{error, info};
+use serde::{Deserialize, Serialize};
use serde_json::Value;
-
+use std::collections::HashMap;
use tokio::{
io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, BufWriter},
process::{ChildStderr, ChildStdin, ChildStdout},
@@ -26,12 +20,11 @@ pub enum Payload {
Response(jsonrpc::Output),
}
-use serde::{Deserialize, Serialize};
/// A type representing all possible values sent from the server to the client.
#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
#[serde(untagged)]
-enum Message {
+enum ServerMessage {
/// A regular JSON-RPC request output (single response).
Output(jsonrpc::Output),
/// A JSON-RPC request or notification.
@@ -39,34 +32,32 @@ enum Message {
}
pub struct Transport {
- incoming: UnboundedSender<jsonrpc::Call>,
- outgoing: UnboundedReceiver<Payload>,
+ client_tx: UnboundedSender<jsonrpc::Call>,
+ client_rx: UnboundedReceiver<Payload>,
pending_requests: HashMap<jsonrpc::Id, Sender<Result<Value>>>,
- headers: HashMap<String, String>,
- writer: BufWriter<ChildStdin>,
- reader: BufReader<ChildStdout>,
- stderr: BufReader<ChildStderr>,
+ server_stdin: BufWriter<ChildStdin>,
+ server_stdout: BufReader<ChildStdout>,
+ server_stderr: BufReader<ChildStderr>,
}
impl Transport {
pub fn start(
- reader: BufReader<ChildStdout>,
- writer: BufWriter<ChildStdin>,
- stderr: BufReader<ChildStderr>,
+ server_stdout: BufReader<ChildStdout>,
+ server_stdin: BufWriter<ChildStdin>,
+ server_stderr: BufReader<ChildStderr>,
) -> (UnboundedReceiver<jsonrpc::Call>, UnboundedSender<Payload>) {
- let (incoming, rx) = unbounded_channel();
- let (tx, outgoing) = unbounded_channel();
+ let (client_tx, rx) = unbounded_channel();
+ let (tx, client_rx) = unbounded_channel();
let transport = Self {
- reader,
- writer,
- stderr,
- incoming,
- outgoing,
+ server_stdout,
+ server_stdin,
+ server_stderr,
+ client_tx,
+ client_rx,
pending_requests: HashMap::default(),
- headers: HashMap::default(),
};
tokio::spawn(transport.duplex());
@@ -74,105 +65,104 @@ impl Transport {
(rx, tx)
}
- async fn recv(
+ async fn recv_server_message(
reader: &mut (impl AsyncBufRead + Unpin + Send),
- headers: &mut HashMap<String, String>,
- ) -> core::result::Result<Message, std::io::Error> {
- // read headers
+ buffer: &mut String,
+ ) -> Result<ServerMessage> {
+ let mut content_length = None;
loop {
- let mut header = String::new();
- // detect pipe closed if 0
- reader.read_line(&mut header).await?;
- let header = header.trim();
+ buffer.truncate(0);
+ reader.read_line(buffer).await?;
+ let header = buffer.trim();
if header.is_empty() {
break;
}
- let parts: Vec<&str> = header.split(": ").collect();
- if parts.len() != 2 {
- return Err(std::io::Error::new(
- std::io::ErrorKind::Other,
- "Failed to parse header",
- ));
+ let mut parts = header.split(": ");
+
+ match (parts.next(), parts.next(), parts.next()) {
+ (Some("Content-Length"), Some(value), None) => {
+ content_length = Some(value.parse().unwrap());
+ }
+ (Some(_), Some(_), None) => {}
+ _ => {
+ return Err(std::io::Error::new(
+ std::io::ErrorKind::Other,
+ "Failed to parse header",
+ )
+ .into());
+ }
}
- headers.insert(parts[0].to_string(), parts[1].to_string());
}
- // find content-length
- let content_length = headers.get("Content-Length").unwrap().parse().unwrap();
+ let content_length = content_length.unwrap();
+ //TODO: reuse vector
let mut content = vec![0; content_length];
reader.read_exact(&mut content).await?;
let msg = String::from_utf8(content).unwrap();
- // read data
-
info!("<- {}", msg);
// try parsing as output (server response) or call (server request)
- let output: serde_json::Result<Message> = serde_json::from_str(&msg);
+ let output: serde_json::Result<ServerMessage> = serde_json::from_str(&msg);
Ok(output?)
}
- async fn err(
+ async fn recv_server_error(
err: &mut (impl AsyncBufRead + Unpin + Send),
- ) -> core::result::Result<(), std::io::Error> {
- let mut line = String::new();
- err.read_line(&mut line).await?;
- error!("err <- {}", line);
+ buffer: &mut String,
+ ) -> Result<()> {
+ buffer.truncate(0);
+ err.read_line(buffer).await?;
+ error!("err <- {}", buffer);
Ok(())
}
- pub async fn send_payload(&mut self, payload: Payload) -> io::Result<()> {
- match payload {
+ async fn send_payload_to_server(&mut self, payload: Payload) -> Result<()> {
+ //TODO: reuse string
+ let json = match payload {
Payload::Request { chan, value } => {
self.pending_requests.insert(value.id.clone(), chan);
-
- let json = serde_json::to_string(&value)?;
- self.send(json).await
- }
- Payload::Notification(value) => {
- let json = serde_json::to_string(&value)?;
- self.send(json).await
+ serde_json::to_string(&value)?
}
- Payload::Response(error) => {
- let json = serde_json::to_string(&error)?;
- self.send(json).await
- }
- }
+ Payload::Notification(value) => serde_json::to_string(&value)?,
+ Payload::Response(error) => serde_json::to_string(&error)?,
+ };
+ self.send_string_to_server(json).await
}
- pub async fn send(&mut self, request: String) -> io::Result<()> {
+ async fn send_string_to_server(&mut self, request: String) -> Result<()> {
info!("-> {}", request);
// send the headers
- self.writer
+ self.server_stdin
.write_all(format!("Content-Length: {}\r\n\r\n", request.len()).as_bytes())
.await?;
// send the body
- self.writer.write_all(request.as_bytes()).await?;
+ self.server_stdin.write_all(request.as_bytes()).await?;
- self.writer.flush().await?;
+ self.server_stdin.flush().await?;
Ok(())
}
- async fn recv_msg(&mut self, msg: Message) -> anyhow::Result<()> {
+ async fn process_server_message(&mut self, msg: ServerMessage) -> Result<()> {
match msg {
- Message::Output(output) => self.recv_response(output).await?,
- Message::Call(call) => {
- self.incoming.send(call).unwrap();
+ ServerMessage::Output(output) => self.process_request_response(output).await?,
+ ServerMessage::Call(call) => {
+ self.client_tx.send(call).unwrap();
// let notification = Notification::parse(&method, params);
}
};
Ok(())
}
- async fn recv_response(&mut self, output: jsonrpc::Output) -> io::Result<()> {
+ async fn process_request_response(&mut self, output: jsonrpc::Output) -> Result<()> {
let (id, result) = match output {
jsonrpc::Output::Success(jsonrpc::Success { id, result, .. }) => {
info!("<- {}", result);
@@ -200,29 +190,33 @@ impl Transport {
Ok(())
}
- pub async fn duplex(mut self) {
+ async fn duplex(mut self) {
+ let mut recv_buffer = String::new();
+ let mut err_buffer = String::new();
loop {
tokio::select! {
// client -> server
- msg = self.outgoing.recv() => {
- if msg.is_none() {
- break;
+ msg = self.client_rx.recv() => {
+ match msg {
+ Some(msg) => {
+ self.send_payload_to_server(msg).await.unwrap()
+ },
+ None => break
}
- let msg = msg.unwrap();
-
- self.send_payload(msg).await.unwrap();
}
- // server <- client
- msg = Self::recv(&mut self.reader, &mut self.headers) => {
- if msg.is_err() {
- error!("err: <- {:?}", msg);
- break;
+ // server -> client
+ msg = Self::recv_server_message(&mut self.server_stdout, &mut recv_buffer) => {
+ match msg {
+ Ok(msg) => {
+ self.process_server_message(msg).await.unwrap();
+ }
+ Err(_) => {
+ error!("err: <- {:?}", msg);
+ break;
+ },
}
- let msg = msg.unwrap();
-
- self.recv_msg(msg).await.unwrap();
}
- _msg = Self::err(&mut self.stderr) => {}
+ _msg = Self::recv_server_error(&mut self.server_stderr, &mut err_buffer) => {}
}
}
}