diff options
Diffstat (limited to 'helix-dap/src/transport.rs')
| -rw-r--r-- | helix-dap/src/transport.rs | 282 |
1 files changed, 282 insertions, 0 deletions
diff --git a/helix-dap/src/transport.rs b/helix-dap/src/transport.rs new file mode 100644 index 00000000..1c004cfe --- /dev/null +++ b/helix-dap/src/transport.rs @@ -0,0 +1,282 @@ +use crate::{Error, Result}; +use anyhow::Context; +use log::error; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::collections::HashMap; +use std::sync::Arc; +use tokio::{ + io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, BufWriter}, + process::{ChildStdin, ChildStdout}, + sync::{ + mpsc::{unbounded_channel, Sender, UnboundedReceiver, UnboundedSender}, + Mutex, + }, +}; + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct Request { + #[serde(skip)] + pub back_ch: Option<Sender<Result<Response>>>, + pub seq: u64, + #[serde(rename = "type")] + pub msg_type: String, + pub command: String, + pub arguments: Option<Value>, +} + +#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)] +pub struct Response { + pub seq: u64, + #[serde(rename = "type")] + pub msg_type: String, + pub request_seq: u64, + pub success: bool, + pub command: String, + pub message: Option<String>, + pub body: Option<Value>, +} + +#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)] +pub struct Event { + pub seq: u64, + #[serde(rename = "type")] + pub msg_type: String, + pub event: String, + pub body: Option<Value>, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +#[serde(untagged)] +pub enum Payload { + // type = "event" + Event(Event), + // type = "response" + Response(Response), + // type = "request" + Request(Request), +} + +#[derive(Debug)] +pub struct Transport { + id: usize, + pending_requests: Mutex<HashMap<u64, Sender<Result<Response>>>>, +} + +impl Transport { + pub fn start( + server_stdout: BufReader<ChildStdout>, + server_stdin: BufWriter<ChildStdin>, + id: usize, + ) -> (UnboundedReceiver<Payload>, UnboundedSender<Request>) { + let (client_tx, rx) = unbounded_channel(); + let (tx, client_rx) = unbounded_channel(); + + let transport = Self { + id, + pending_requests: Mutex::new(HashMap::default()), + }; + + let transport = Arc::new(transport); + + tokio::spawn(Self::recv(transport.clone(), server_stdout, client_tx)); + tokio::spawn(Self::send(transport, server_stdin, client_rx)); + + (rx, tx) + } + + async fn recv_server_message( + reader: &mut (impl AsyncBufRead + Unpin + Send), + buffer: &mut String, + ) -> Result<Payload> { + let mut content_length = None; + loop { + buffer.truncate(0); + reader.read_line(buffer).await?; + let header = buffer.trim(); + + if header.is_empty() { + break; + } + + let mut parts = header.split(": "); + + match (parts.next(), parts.next(), parts.next()) { + (Some("Content-Length"), Some(value), None) => { + content_length = Some(value.parse().context("invalid content length")?); + } + (Some(_), Some(_), None) => {} + _ => { + return Err(std::io::Error::new( + std::io::ErrorKind::Other, + "Failed to parse header", + ) + .into()); + } + } + } + + let content_length = content_length.context("missing content length")?; + + //TODO: reuse vector + let mut content = vec![0; content_length]; + reader.read_exact(&mut content).await?; + let msg = std::str::from_utf8(&content).context("invalid utf8 from server")?; + + // TODO: `info!` here + println!("<- DAP {}", msg); + + // try parsing as output (server response) or call (server request) + let output: serde_json::Result<Payload> = serde_json::from_str(msg); + + Ok(output?) + } + + async fn send_payload_to_server( + &self, + server_stdin: &mut BufWriter<ChildStdin>, + req: Request, + ) -> Result<()> { + let json = serde_json::to_string(&req)?; + match req.back_ch { + Some(back) => { + self.pending_requests.lock().await.insert(req.seq, back); + () + } + None => {} + } + self.send_string_to_server(server_stdin, json).await + } + + async fn send_string_to_server( + &self, + server_stdin: &mut BufWriter<ChildStdin>, + request: String, + ) -> Result<()> { + // TODO: `info!` here + println!("-> DAP {}", request); + + // send the headers + server_stdin + .write_all(format!("Content-Length: {}\r\n\r\n", request.len()).as_bytes()) + .await?; + + // send the body + server_stdin.write_all(request.as_bytes()).await?; + + server_stdin.flush().await?; + + Ok(()) + } + + async fn process_server_message( + &self, + client_tx: &UnboundedSender<Payload>, + msg: Payload, + ) -> Result<()> { + let (id, result) = match msg { + Payload::Response(Response { + success: true, + seq, + request_seq, + .. + }) => { + // TODO: `info!` here + println!("<- DAP success ({}, in response to {})", seq, request_seq); + if let Payload::Response(val) = msg { + (request_seq, Ok(val)) + } else { + unreachable!(); + } + } + Payload::Response(Response { + success: false, + message, + body, + request_seq, + command, + .. + }) => { + // TODO: `error!` here + println!( + "<- DAP error {:?} ({:?}) for command #{} {}", + message, body, request_seq, command + ); + ( + request_seq, + Err(Error::Other(anyhow::format_err!("{:?}", body))), + ) + } + Payload::Request(Request { + ref command, + ref seq, + .. + }) => { + // TODO: `info!` here + println!("<- DAP request {} #{}", command, seq); + client_tx.send(msg).expect("Failed to send"); + return Ok(()); + } + Payload::Event(Event { + ref event, ref seq, .. + }) => { + // TODO: `info!` here + println!("<- DAP event {} #{}", event, seq); + client_tx.send(msg).expect("Failed to send"); + return Ok(()); + } + }; + + let tx = self + .pending_requests + .lock() + .await + .remove(&id) + .expect("pending_request with id not found!"); + + match tx.send(result).await { + Ok(_) => (), + Err(_) => error!( + "Tried sending response into a closed channel (id={:?}), original request likely timed out", + id + ), + }; + + Ok(()) + } + + async fn recv( + transport: Arc<Self>, + mut server_stdout: BufReader<ChildStdout>, + client_tx: UnboundedSender<Payload>, + ) { + let mut recv_buffer = String::new(); + loop { + match Self::recv_server_message(&mut server_stdout, &mut recv_buffer).await { + Ok(msg) => { + transport + .process_server_message(&client_tx, msg) + .await + .unwrap(); + } + Err(err) => { + error!("err: <- {:?}", err); + break; + } + } + } + } + + async fn send( + transport: Arc<Self>, + mut server_stdin: BufWriter<ChildStdin>, + mut client_rx: UnboundedReceiver<Request>, + ) { + while let Some(req) = client_rx.recv().await { + transport + .send_payload_to_server(&mut server_stdin, req) + .await + .unwrap() + } + } +} |