@@ -1,272 +1,271 @@
1−use crate::{Error, Result};
2−use anyhow::Context;
3−use log::{error, info};
4−use serde::{Deserialize, Serialize};
5−use serde_json::Value;
6−use std::collections::HashMap;
7−use std::sync::Arc;
8−use tokio::{
9− io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, BufWriter},
10− process::{ChildStdin, ChildStdout},
11− sync::{
12− mpsc::{unbounded_channel, Sender, UnboundedReceiver, UnboundedSender},
13− Mutex,
14− },
15−};
16−
17−#[derive(Debug, Clone, Deserialize, Serialize)]
18−pub struct Request {
19− #[serde(skip)]
20− pub back_ch: Option<Sender<Result<Response>>>,
21− pub seq: u64,
22− #[serde(rename = "type")]
23− pub msg_type: String,
24− pub command: String,
25− pub arguments: Option<Value>,
26−}
27−
28−#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
29−pub struct Response {
30− pub seq: u64,
31− #[serde(rename = "type")]
32− pub msg_type: String,
33− pub request_seq: u64,
34− pub success: bool,
35− pub command: String,
36− pub message: Option<String>,
37− pub body: Option<Value>,
38−}
39−
40−#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
41−pub struct Event {
42− pub seq: u64,
43− #[serde(rename = "type")]
44− pub msg_type: String,
45− pub event: String,
46− pub body: Option<Value>,
47−}
48−
49−#[derive(Debug, Clone, Deserialize, Serialize)]
50−#[serde(untagged)]
51−pub enum Payload {
52− // type = "event"
53− Event(Event),
54− // type = "response"
55− Response(Response),
56− // type = "request"
57− Request(Request),
58−}
59−
60−#[derive(Debug)]
61−pub struct Transport {
62− id: usize,
63− pending_requests: Mutex<HashMap<u64, Sender<Result<Response>>>>,
64−}
65−
66−impl Transport {
67− pub fn start(
68− server_stdout: BufReader<ChildStdout>,
69− server_stdin: BufWriter<ChildStdin>,
70− id: usize,
71− ) -> (UnboundedReceiver<Payload>, UnboundedSender<Request>) {
72− let (client_tx, rx) = unbounded_channel();
73− let (tx, client_rx) = unbounded_channel();
74−
75− let transport = Self {
76− id,
77− pending_requests: Mutex::new(HashMap::default()),
78− };
79−
80− let transport = Arc::new(transport);
81−
82− tokio::spawn(Self::recv(transport.clone(), server_stdout, client_tx));
83− tokio::spawn(Self::send(transport, server_stdin, client_rx));
84−
85− (rx, tx)
86− }
87−
88− async fn recv_server_message(
89− reader: &mut (impl AsyncBufRead + Unpin + Send),
90− buffer: &mut String,
91− ) -> Result<Payload> {
92− let mut content_length = None;
93− loop {
94− buffer.truncate(0);
95− reader.read_line(buffer).await?;
96− let header = buffer.trim();
97−
98− if header.is_empty() {
99− break;
100− }
101−
102− let mut parts = header.split(": ");
103−
104− match (parts.next(), parts.next(), parts.next()) {
105− (Some("Content-Length"), Some(value), None) => {
106− content_length = Some(value.parse().context("invalid content length")?);
107− }
108− (Some(_), Some(_), None) => {}
109− _ => {
110− return Err(std::io::Error::new(
111− std::io::ErrorKind::Other,
112− "Failed to parse header",
113− )
114− .into());
115− }
116− }
117− }
118−
119− let content_length = content_length.context("missing content length")?;
120−
121− //TODO: reuse vector
122− let mut content = vec![0; content_length];
123− reader.read_exact(&mut content).await?;
124− let msg = std::str::from_utf8(&content).context("invalid utf8 from server")?;
125−
126− info!("<- DAP {}", msg);
127−
128− // try parsing as output (server response) or call (server request)
129− let output: serde_json::Result<Payload> = serde_json::from_str(msg);
130−
131− Ok(output?)
132− }
133−
134− async fn send_payload_to_server(
135− &self,
136− server_stdin: &mut BufWriter<ChildStdin>,
137− req: Request,
138− ) -> Result<()> {
139− let json = serde_json::to_string(&req)?;
140− if let Some(back) = req.back_ch {
141− self.pending_requests.lock().await.insert(req.seq, back);
142− }
143− self.send_string_to_server(server_stdin, json).await
144− }
145−
146− async fn send_string_to_server(
147− &self,
148− server_stdin: &mut BufWriter<ChildStdin>,
149− request: String,
150− ) -> Result<()> {
151− info!("-> DAP {}", request);
152−
153− // send the headers
154− server_stdin
155− .write_all(format!("Content-Length: {}\r\n\r\n", request.len()).as_bytes())
156− .await?;
157−
158− // send the body
159− server_stdin.write_all(request.as_bytes()).await?;
160−
161− server_stdin.flush().await?;
162−
163− Ok(())
164− }
165−
166− async fn process_server_message(
167− &self,
168− client_tx: &UnboundedSender<Payload>,
169− msg: Payload,
170− ) -> Result<()> {
171− let (id, result) = match msg {
172− Payload::Response(Response {
173− success: true,
174− seq,
175− request_seq,
176− ..
177− }) => {
178− info!("<- DAP success ({}, in response to {})", seq, request_seq);
179− if let Payload::Response(val) = msg {
180− (request_seq, Ok(val))
181− } else {
182− unreachable!();
183− }
184− }
185− Payload::Response(Response {
186− success: false,
187− message,
188− body,
189− request_seq,
190− command,
191− ..
192− }) => {
193− error!(
194− "<- DAP error {:?} ({:?}) for command #{} {}",
195− message, body, request_seq, command
196− );
197− (
198− request_seq,
199− Err(Error::Other(anyhow::format_err!("{:?}", body))),
200− )
201− }
202− Payload::Request(Request {
203− ref command,
204− ref seq,
205− ..
206− }) => {
207− info!("<- DAP request {} #{}", command, seq);
208− client_tx.send(msg).expect("Failed to send");
209− return Ok(());
210− }
211− Payload::Event(Event {
212− ref event, ref seq, ..
213− }) => {
214− info!("<- DAP event {} #{}", event, seq);
215− client_tx.send(msg).expect("Failed to send");
216− return Ok(());
217− }
218− };
219−
220− let tx = self
221− .pending_requests
222− .lock()
223− .await
224− .remove(&id)
225− .expect("pending_request with id not found!");
226−
227− match tx.send(result).await {
228− Ok(_) => (),
229− Err(_) => error!(
230− "Tried sending response into a closed channel (id={:?}), original request likely timed out",
231− id
232− ),
233− };
234−
235− Ok(())
236− }
237−
238− async fn recv(
239− transport: Arc<Self>,
240− mut server_stdout: BufReader<ChildStdout>,
241− client_tx: UnboundedSender<Payload>,
242− ) {
243− let mut recv_buffer = String::new();
244− loop {
245− match Self::recv_server_message(&mut server_stdout, &mut recv_buffer).await {
246− Ok(msg) => {
247− transport
248− .process_server_message(&client_tx, msg)
249− .await
250− .unwrap();
251− }
252− Err(err) => {
253− error!("err: <- {:?}", err);
254− break;
255− }
256− }
257− }
258− }
259−
260− async fn send(
261− transport: Arc<Self>,
262− mut server_stdin: BufWriter<ChildStdin>,
263− mut client_rx: UnboundedReceiver<Request>,
264− ) {
265− while let Some(req) = client_rx.recv().await {
266− transport
267− .send_payload_to_server(&mut server_stdin, req)
268− .await
269− .unwrap()
270− }
271− }
272−}
1+use crate::{Error, Result};
2+use anyhow::Context;
3+use log::{error, info};
4+use serde::{Deserialize, Serialize};
5+use serde_json::Value;
6+use std::collections::HashMap;
7+use std::sync::Arc;
8+use tokio::{
9+ io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWrite, AsyncWriteExt},
10+ sync::{
11+ mpsc::{unbounded_channel, Sender, UnboundedReceiver, UnboundedSender},
12+ Mutex,
13+ },
14+};
15+
16+#[derive(Debug, Clone, Deserialize, Serialize)]
17+pub struct Request {
18+ #[serde(skip)]
19+ pub back_ch: Option<Sender<Result<Response>>>,
20+ pub seq: u64,
21+ #[serde(rename = "type")]
22+ pub msg_type: String,
23+ pub command: String,
24+ pub arguments: Option<Value>,
25+}
26+
27+#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
28+pub struct Response {
29+ pub seq: u64,
30+ #[serde(rename = "type")]
31+ pub msg_type: String,
32+ pub request_seq: u64,
33+ pub success: bool,
34+ pub command: String,
35+ pub message: Option<String>,
36+ pub body: Option<Value>,
37+}
38+
39+#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
40+pub struct Event {
41+ pub seq: u64,
42+ #[serde(rename = "type")]
43+ pub msg_type: String,
44+ pub event: String,
45+ pub body: Option<Value>,
46+}
47+
48+#[derive(Debug, Clone, Deserialize, Serialize)]
49+#[serde(untagged)]
50+pub enum Payload {
51+ // type = "event"
52+ Event(Event),
53+ // type = "response"
54+ Response(Response),
55+ // type = "request"
56+ Request(Request),
57+}
58+
59+#[derive(Debug)]
60+pub struct Transport {
61+ id: usize,
62+ pending_requests: Mutex<HashMap<u64, Sender<Result<Response>>>>,
63+}
64+
65+impl Transport {
66+ pub fn start(
67+ server_stdout: Box<dyn AsyncBufRead + Unpin + Send>,
68+ server_stdin: Box<dyn AsyncWrite + Unpin + Send>,
69+ id: usize,
70+ ) -> (UnboundedReceiver<Payload>, UnboundedSender<Request>) {
71+ let (client_tx, rx) = unbounded_channel();
72+ let (tx, client_rx) = unbounded_channel();
73+
74+ let transport = Self {
75+ id,
76+ pending_requests: Mutex::new(HashMap::default()),
77+ };
78+
79+ let transport = Arc::new(transport);
80+
81+ tokio::spawn(Self::recv(transport.clone(), server_stdout, client_tx));
82+ tokio::spawn(Self::send(transport, server_stdin, client_rx));
83+
84+ (rx, tx)
85+ }
86+
87+ async fn recv_server_message(
88+ reader: &mut Box<dyn AsyncBufRead + Unpin + Send>,
89+ buffer: &mut String,
90+ ) -> Result<Payload> {
91+ let mut content_length = None;
92+ loop {
93+ buffer.truncate(0);
94+ reader.read_line(buffer).await?;
95+ let header = buffer.trim();
96+
97+ if header.is_empty() {
98+ break;
99+ }
100+
101+ let mut parts = header.split(": ");
102+
103+ match (parts.next(), parts.next(), parts.next()) {
104+ (Some("Content-Length"), Some(value), None) => {
105+ content_length = Some(value.parse().context("invalid content length")?);
106+ }
107+ (Some(_), Some(_), None) => {}
108+ _ => {
109+ return Err(std::io::Error::new(
110+ std::io::ErrorKind::Other,
111+ "Failed to parse header",
112+ )
113+ .into());
114+ }
115+ }
116+ }
117+
118+ let content_length = content_length.context("missing content length")?;
119+
120+ //TODO: reuse vector
121+ let mut content = vec![0; content_length];
122+ reader.read_exact(&mut content).await?;
123+ let msg = std::str::from_utf8(&content).context("invalid utf8 from server")?;
124+
125+ info!("<- DAP {}", msg);
126+
127+ // try parsing as output (server response) or call (server request)
128+ let output: serde_json::Result<Payload> = serde_json::from_str(msg);
129+
130+ Ok(output?)
131+ }
132+
133+ async fn send_payload_to_server(
134+ &self,
135+ server_stdin: &mut Box<dyn AsyncWrite + Unpin + Send>,
136+ req: Request,
137+ ) -> Result<()> {
138+ let json = serde_json::to_string(&req)?;
139+ if let Some(back) = req.back_ch {
140+ self.pending_requests.lock().await.insert(req.seq, back);
141+ }
142+ self.send_string_to_server(server_stdin, json).await
143+ }
144+
145+ async fn send_string_to_server(
146+ &self,
147+ server_stdin: &mut Box<dyn AsyncWrite + Unpin + Send>,
148+ request: String,
149+ ) -> Result<()> {
150+ info!("-> DAP {}", request);
151+
152+ // send the headers
153+ server_stdin
154+ .write_all(format!("Content-Length: {}\r\n\r\n", request.len()).as_bytes())
155+ .await?;
156+
157+ // send the body
158+ server_stdin.write_all(request.as_bytes()).await?;
159+
160+ server_stdin.flush().await?;
161+
162+ Ok(())
163+ }
164+
165+ async fn process_server_message(
166+ &self,
167+ client_tx: &UnboundedSender<Payload>,
168+ msg: Payload,
169+ ) -> Result<()> {
170+ let (id, result) = match msg {
171+ Payload::Response(Response {
172+ success: true,
173+ seq,
174+ request_seq,
175+ ..
176+ }) => {
177+ info!("<- DAP success ({}, in response to {})", seq, request_seq);
178+ if let Payload::Response(val) = msg {
179+ (request_seq, Ok(val))
180+ } else {
181+ unreachable!();
182+ }
183+ }
184+ Payload::Response(Response {
185+ success: false,
186+ message,
187+ body,
188+ request_seq,
189+ command,
190+ ..
191+ }) => {
192+ error!(
193+ "<- DAP error {:?} ({:?}) for command #{} {}",
194+ message, body, request_seq, command
195+ );
196+ (
197+ request_seq,
198+ Err(Error::Other(anyhow::format_err!("{:?}", body))),
199+ )
200+ }
201+ Payload::Request(Request {
202+ ref command,
203+ ref seq,
204+ ..
205+ }) => {
206+ info!("<- DAP request {} #{}", command, seq);
207+ client_tx.send(msg).expect("Failed to send");
208+ return Ok(());
209+ }
210+ Payload::Event(Event {
211+ ref event, ref seq, ..
212+ }) => {
213+ info!("<- DAP event {} #{}", event, seq);
214+ client_tx.send(msg).expect("Failed to send");
215+ return Ok(());
216+ }
217+ };
218+
219+ let tx = self
220+ .pending_requests
221+ .lock()
222+ .await
223+ .remove(&id)
224+ .expect("pending_request with id not found!");
225+
226+ match tx.send(result).await {
227+ Ok(_) => (),
228+ Err(_) => error!(
229+ "Tried sending response into a closed channel (id={:?}), original request likely timed out",
230+ id
231+ ),
232+ };
233+
234+ Ok(())
235+ }
236+
237+ async fn recv(
238+ transport: Arc<Self>,
239+ mut server_stdout: Box<dyn AsyncBufRead + Unpin + Send>,
240+ client_tx: UnboundedSender<Payload>,
241+ ) {
242+ let mut recv_buffer = String::new();
243+ loop {
244+ match Self::recv_server_message(&mut server_stdout, &mut recv_buffer).await {
245+ Ok(msg) => {
246+ transport
247+ .process_server_message(&client_tx, msg)
248+ .await
249+ .unwrap();
250+ }
251+ Err(err) => {
252+ error!("err: <- {:?}", err);
253+ break;
254+ }
255+ }
256+ }
257+ }
258+
259+ async fn send(
260+ transport: Arc<Self>,
261+ mut server_stdin: Box<dyn AsyncWrite + Unpin + Send>,
262+ mut client_rx: UnboundedReceiver<Request>,
263+ ) {
264+ while let Some(req) = client_rx.recv().await {
265+ transport
266+ .send_payload_to_server(&mut server_stdin, req)
267+ .await
268+ .unwrap()
269+ }
270+ }
271+}