git.iko.soy/helix
on 2158366 help sign in
helix/helix-lsp/src/transport.rs
263 lines 7.9 kB history raw
1use crate::Result;
2use anyhow::Context;
3use jsonrpc_core as jsonrpc;
4use log::{error, info};
5use serde::{Deserialize, Serialize};
6use serde_json::Value;
7use std::collections::HashMap;
8use std::sync::Arc;
9use tokio::{
10 io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, BufWriter},
11 process::{ChildStderr, ChildStdin, ChildStdout},
12 sync::{
13 mpsc::{unbounded_channel, Sender, UnboundedReceiver, UnboundedSender},
14 Mutex,
15 },
16};
18#[derive(Debug)]
19pub enum Payload {
20 Request {
21 chan: Sender<Result<Value>>,
22 value: jsonrpc::MethodCall,
23 },
24 Notification(jsonrpc::Notification),
25 Response(jsonrpc::Output),
28/// A type representing all possible values sent from the server to the client.
29#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
30#[serde(deny_unknown_fields)]
31#[serde(untagged)]
32enum ServerMessage {
33 /// A regular JSON-RPC request output (single response).
34 Output(jsonrpc::Output),
35 /// A JSON-RPC request or notification.
36 Call(jsonrpc::Call),
39#[derive(Debug)]
40pub struct Transport {
41 id: usize,
42 pending_requests: Mutex<HashMap<jsonrpc::Id, Sender<Result<Value>>>>,
45impl Transport {
46 pub fn start(
47 server_stdout: BufReader<ChildStdout>,
48 server_stdin: BufWriter<ChildStdin>,
49 server_stderr: BufReader<ChildStderr>,
50 id: usize,
51 ) -> (
52 UnboundedReceiver<(usize, jsonrpc::Call)>,
53 UnboundedSender<Payload>,
54 ) {
55 let (client_tx, rx) = unbounded_channel();
56 let (tx, client_rx) = unbounded_channel();
58 let transport = Self {
59 id,
60 pending_requests: Mutex::new(HashMap::default()),
61 };
63 let transport = Arc::new(transport);
65 tokio::spawn(Self::recv(transport.clone(), server_stdout, client_tx));
66 tokio::spawn(Self::err(transport.clone(), server_stderr));
67 tokio::spawn(Self::send(transport, server_stdin, client_rx));
69 (rx, tx)
70 }
72 async fn recv_server_message(
73 reader: &mut (impl AsyncBufRead + Unpin + Send),
74 buffer: &mut String,
75 ) -> Result<ServerMessage> {
76 let mut content_length = None;
77 loop {
78 buffer.truncate(0);
79 reader.read_line(buffer).await?;
80 let header = buffer.trim();
82 if header.is_empty() {
83 break;
84 }
86 let mut parts = header.split(": ");
88 match (parts.next(), parts.next(), parts.next()) {
89 (Some("Content-Length"), Some(value), None) => {
90 content_length = Some(value.parse().context("invalid content length")?);
91 }
92 (Some(_), Some(_), None) => {}
93 _ => {
94 return Err(std::io::Error::new(
95 std::io::ErrorKind::Other,
96 "Failed to parse header",
97 )
98 .into());
99 }
100 }
101 }
103 let content_length = content_length.context("missing content length")?;
105 //TODO: reuse vector
106 let mut content = vec![0; content_length];
107 reader.read_exact(&mut content).await?;
108 let msg = std::str::from_utf8(&content).context("invalid utf8 from server")?;
110 info!("<- {}", msg);
112 // try parsing as output (server response) or call (server request)
113 let output: serde_json::Result<ServerMessage> = serde_json::from_str(msg);
115 Ok(output?)
116 }
118 async fn recv_server_error(
119 err: &mut (impl AsyncBufRead + Unpin + Send),
120 buffer: &mut String,
121 ) -> Result<()> {
122 buffer.truncate(0);
123 err.read_line(buffer).await?;
124 error!("err <- {}", buffer);
126 Ok(())
127 }
129 async fn send_payload_to_server(
130 &self,
131 server_stdin: &mut BufWriter<ChildStdin>,
132 payload: Payload,
133 ) -> Result<()> {
134 //TODO: reuse string
135 let json = match payload {
136 Payload::Request { chan, value } => {
137 self.pending_requests
138 .lock()
139 .await
140 .insert(value.id.clone(), chan);
141 serde_json::to_string(&value)?
142 }
143 Payload::Notification(value) => serde_json::to_string(&value)?,
144 Payload::Response(error) => serde_json::to_string(&error)?,
145 };
146 self.send_string_to_server(server_stdin, json).await
147 }
149 async fn send_string_to_server(
150 &self,
151 server_stdin: &mut BufWriter<ChildStdin>,
152 request: String,
153 ) -> Result<()> {
154 info!("-> {}", request);
156 // send the headers
157 server_stdin
158 .write_all(format!("Content-Length: {}\r\n\r\n", request.len()).as_bytes())
159 .await?;
161 // send the body
162 server_stdin.write_all(request.as_bytes()).await?;
164 server_stdin.flush().await?;
166 Ok(())
167 }
169 async fn process_server_message(
170 &self,
171 client_tx: &UnboundedSender<(usize, jsonrpc::Call)>,
172 msg: ServerMessage,
173 ) -> Result<()> {
174 match msg {
175 ServerMessage::Output(output) => self.process_request_response(output).await?,
176 ServerMessage::Call(call) => {
177 client_tx
178 .send((self.id, call))
179 .context("failed to send a message to server")?;
180 // let notification = Notification::parse(&method, params);
181 }
182 };
183 Ok(())
184 }
186 async fn process_request_response(&self, output: jsonrpc::Output) -> Result<()> {
187 let (id, result) = match output {
188 jsonrpc::Output::Success(jsonrpc::Success { id, result, .. }) => {
189 info!("<- {}", result);
190 (id, Ok(result))
191 }
192 jsonrpc::Output::Failure(jsonrpc::Failure { id, error, .. }) => {
193 error!("<- {}", error);
194 (id, Err(error.into()))
195 }
196 };
198 let tx = self
199 .pending_requests
200 .lock()
201 .await
202 .remove(&id)
203 .expect("pending_request with id not found!");
205 match tx.send(result).await {
206 Ok(_) => (),
207 Err(_) => error!(
208 "Tried sending response into a closed channel (id={:?}), original request likely timed out",
209 id
210 ),
211 };
213 Ok(())
214 }
216 async fn recv(
217 transport: Arc<Self>,
218 mut server_stdout: BufReader<ChildStdout>,
219 client_tx: UnboundedSender<(usize, jsonrpc::Call)>,
220 ) {
221 let mut recv_buffer = String::new();
222 loop {
223 match Self::recv_server_message(&mut server_stdout, &mut recv_buffer).await {
224 Ok(msg) => {
225 transport
226 .process_server_message(&client_tx, msg)
227 .await
228 .unwrap();
229 }
230 Err(err) => {
231 error!("err: <- {:?}", err);
232 break;
233 }
234 }
235 }
236 }
238 async fn err(_transport: Arc<Self>, mut server_stderr: BufReader<ChildStderr>) {
239 let mut recv_buffer = String::new();
240 loop {
241 match Self::recv_server_error(&mut server_stderr, &mut recv_buffer).await {
242 Ok(_) => {}
243 Err(err) => {
244 error!("err: <- {:?}", err);
245 break;
246 }
247 }
248 }
249 }
251 async fn send(
252 transport: Arc<Self>,
253 mut server_stdin: BufWriter<ChildStdin>,
254 mut client_rx: UnboundedReceiver<Payload>,
255 ) {
256 while let Some(msg) = client_rx.recv().await {
257 transport
258 .send_payload_to_server(&mut server_stdin, msg)
259 .await
260 .unwrap()
261 }
262 }