helix-lsp cleanup

1 files changed, 87 insertions, 93 deletions

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) => {}
             }
         }
     }