git.iko.soy/helix
on 2158366 help sign in
helix/helix-dap/src/transport.rs
246 lines 7.5 kB history raw
1use crate::{Error, Event, Result};
2use anyhow::Context;
3use log::{error, info, warn};
4use serde::{Deserialize, Serialize};
5use serde_json::Value;
6use std::collections::HashMap;
7use std::sync::Arc;
8use tokio::{
9 io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWrite, AsyncWriteExt},
10 sync::{
11 mpsc::{unbounded_channel, Sender, UnboundedReceiver, UnboundedSender},
12 Mutex,
13 },
14};
16#[derive(Debug, Clone, Deserialize, Serialize)]
17pub struct Request {
18 #[serde(skip)]
19 pub back_ch: Option<Sender<Result<Response>>>,
20 pub seq: u64,
21 pub command: String,
22 pub arguments: Option<Value>,
25#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
26pub struct Response {
27 // seq is omitted as unused and is not sent by some implementations
28 pub request_seq: u64,
29 pub success: bool,
30 pub command: String,
31 pub message: Option<String>,
32 pub body: Option<Value>,
35#[derive(Debug, Clone, Deserialize, Serialize)]
36#[serde(tag = "type", rename_all = "camelCase")]
37pub enum Payload {
38 // type = "event"
39 Event(Event),
40 // type = "response"
41 Response(Response),
42 // type = "request"
43 Request(Request),
46#[derive(Debug)]
47pub struct Transport {
48 id: usize,
49 pending_requests: Mutex<HashMap<u64, Sender<Result<Response>>>>,
52impl Transport {
53 pub fn start(
54 server_stdout: Box<dyn AsyncBufRead + Unpin + Send>,
55 server_stdin: Box<dyn AsyncWrite + Unpin + Send>,
56 id: usize,
57 ) -> (UnboundedReceiver<Payload>, UnboundedSender<Request>) {
58 let (client_tx, rx) = unbounded_channel();
59 let (tx, client_rx) = unbounded_channel();
61 let transport = Self {
62 id,
63 pending_requests: Mutex::new(HashMap::default()),
64 };
66 let transport = Arc::new(transport);
68 tokio::spawn(Self::recv(transport.clone(), server_stdout, client_tx));
69 tokio::spawn(Self::send(transport, server_stdin, client_rx));
71 (rx, tx)
72 }
74 async fn recv_server_message(
75 reader: &mut Box<dyn AsyncBufRead + Unpin + Send>,
76 buffer: &mut String,
77 ) -> Result<Payload> {
78 let mut content_length = None;
79 loop {
80 buffer.truncate(0);
81 reader.read_line(buffer).await?;
82 let header = buffer.trim();
84 if header.is_empty() {
85 break;
86 }
88 let mut parts = header.split(": ");
90 match (parts.next(), parts.next(), parts.next()) {
91 (Some("Content-Length"), Some(value), None) => {
92 content_length = Some(value.parse().context("invalid content length")?);
93 }
94 (Some(_), Some(_), None) => {}
95 _ => {
96 return Err(std::io::Error::new(
97 std::io::ErrorKind::Other,
98 "Failed to parse header",
99 )
100 .into());
101 }
102 }
103 }
105 let content_length = content_length.context("missing content length")?;
107 //TODO: reuse vector
108 let mut content = vec![0; content_length];
109 reader.read_exact(&mut content).await?;
110 let msg = std::str::from_utf8(&content).context("invalid utf8 from server")?;
112 info!("<- DAP {}", msg);
114 // try parsing as output (server response) or call (server request)
115 let output: serde_json::Result<Payload> = serde_json::from_str(msg);
117 Ok(output?)
118 }
120 async fn send_payload_to_server(
121 &self,
122 server_stdin: &mut Box<dyn AsyncWrite + Unpin + Send>,
123 mut req: Request,
124 ) -> Result<()> {
125 let back_ch = req.back_ch.take();
126 let seq = req.seq;
127 let json = serde_json::to_string(&Payload::Request(req))?;
128 if let Some(back) = back_ch {
129 self.pending_requests.lock().await.insert(seq, back);
130 }
131 self.send_string_to_server(server_stdin, json).await
132 }
134 async fn send_string_to_server(
135 &self,
136 server_stdin: &mut Box<dyn AsyncWrite + Unpin + Send>,
137 request: String,
138 ) -> Result<()> {
139 info!("-> DAP {}", request);
141 // send the headers
142 server_stdin
143 .write_all(format!("Content-Length: {}\r\n\r\n", request.len()).as_bytes())
144 .await?;
146 // send the body
147 server_stdin.write_all(request.as_bytes()).await?;
149 server_stdin.flush().await?;
151 Ok(())
152 }
154 fn process_response(res: Response) -> Result<Response> {
155 if res.success {
156 info!("<- DAP success in response to {}", res.request_seq);
158 Ok(res)
159 } else {
160 error!(
161 "<- DAP error {:?} ({:?}) for command #{} {}",
162 res.message, res.body, res.request_seq, res.command
163 );
165 Err(Error::Other(anyhow::format_err!("{:?}", res.body)))
166 }
167 }
169 async fn process_server_message(
170 &self,
171 client_tx: &UnboundedSender<Payload>,
172 msg: Payload,
173 ) -> Result<()> {
174 match msg {
175 Payload::Response(res) => {
176 let request_seq = res.request_seq;
177 let tx = self.pending_requests.lock().await.remove(&request_seq);
179 match tx {
180 Some(tx) => match tx.send(Self::process_response(res)).await {
181 Ok(_) => (),
182 Err(_) => error!(
183 "Tried sending response into a closed channel (id={:?}), original request likely timed out",
184 request_seq
185 ),
186 }
187 None => {
188 warn!("Response to nonexistent request #{}", res.request_seq);
189 client_tx.send(Payload::Response(res)).expect("Failed to send");
190 }
191 }
193 Ok(())
194 }
195 Payload::Request(Request {
196 ref command,
197 ref seq,
198 ..
199 }) => {
200 info!("<- DAP request {} #{}", command, seq);
201 client_tx.send(msg).expect("Failed to send");
202 Ok(())
203 }
204 Payload::Event(ref event) => {
205 info!("<- DAP event {:?}", event);
206 client_tx.send(msg).expect("Failed to send");
207 Ok(())
208 }
209 }
210 }
212 async fn recv(
213 transport: Arc<Self>,
214 mut server_stdout: Box<dyn AsyncBufRead + Unpin + Send>,
215 client_tx: UnboundedSender<Payload>,
216 ) {
217 let mut recv_buffer = String::new();
218 loop {
219 match Self::recv_server_message(&mut server_stdout, &mut recv_buffer).await {
220 Ok(msg) => {
221 transport
222 .process_server_message(&client_tx, msg)
223 .await
224 .unwrap();
225 }
226 Err(err) => {
227 error!("err: <- {:?}", err);
228 break;
229 }
230 }
231 }
232 }
234 async fn send(
235 transport: Arc<Self>,
236 mut server_stdin: Box<dyn AsyncWrite + Unpin + Send>,
237 mut client_rx: UnboundedReceiver<Request>,
238 ) {
239 while let Some(req) = client_rx.recv().await {
240 transport
241 .send_payload_to_server(&mut server_stdin, req)
242 .await
243 .unwrap()
244 }
245 }