@@ -1,276 +1,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, 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− match req.back_ch {
141− Some(back) => {
142− self.pending_requests.lock().await.insert(req.seq, back);
143− ()
144− }
145− None => {}
146− }
147− self.send_string_to_server(server_stdin, json).await
148− }
149−
150− async fn send_string_to_server(
151− &self,
152− server_stdin: &mut BufWriter<ChildStdin>,
153− request: String,
154− ) -> Result<()> {
155− info!("-> DAP {}", request);
156−
157− // send the headers
158− server_stdin
159− .write_all(format!("Content-Length: {}\r\n\r\n", request.len()).as_bytes())
160− .await?;
161−
162− // send the body
163− server_stdin.write_all(request.as_bytes()).await?;
164−
165− server_stdin.flush().await?;
166−
167− Ok(())
168− }
169−
170− async fn process_server_message(
171− &self,
172− client_tx: &UnboundedSender<Payload>,
173− msg: Payload,
174− ) -> Result<()> {
175− let (id, result) = match msg {
176− Payload::Response(Response {
177− success: true,
178− seq,
179− request_seq,
180− ..
181− }) => {
182− info!("<- DAP success ({}, in response to {})", seq, request_seq);
183− if let Payload::Response(val) = msg {
184− (request_seq, Ok(val))
185− } else {
186− unreachable!();
187− }
188− }
189− Payload::Response(Response {
190− success: false,
191− message,
192− body,
193− request_seq,
194− command,
195− ..
196− }) => {
197− error!(
198− "<- DAP error {:?} ({:?}) for command #{} {}",
199− message, body, request_seq, command
200− );
201− (
202− request_seq,
203− Err(Error::Other(anyhow::format_err!("{:?}", body))),
204− )
205− }
206− Payload::Request(Request {
207− ref command,
208− ref seq,
209− ..
210− }) => {
211− info!("<- DAP request {} #{}", command, seq);
212− client_tx.send(msg).expect("Failed to send");
213− return Ok(());
214− }
215− Payload::Event(Event {
216− ref event, ref seq, ..
217− }) => {
218− info!("<- DAP event {} #{}", event, seq);
219− client_tx.send(msg).expect("Failed to send");
220− return Ok(());
221− }
222− };
223−
224− let tx = self
225− .pending_requests
226− .lock()
227− .await
228− .remove(&id)
229− .expect("pending_request with id not found!");
230−
231− match tx.send(result).await {
232− Ok(_) => (),
233− Err(_) => error!(
234− "Tried sending response into a closed channel (id={:?}), original request likely timed out",
235− id
236− ),
237− };
238−
239− Ok(())
240− }
241−
242− async fn recv(
243− transport: Arc<Self>,
244− mut server_stdout: BufReader<ChildStdout>,
245− client_tx: UnboundedSender<Payload>,
246− ) {
247− let mut recv_buffer = String::new();
248− loop {
249− match Self::recv_server_message(&mut server_stdout, &mut recv_buffer).await {
250− Ok(msg) => {
251− transport
252− .process_server_message(&client_tx, msg)
253− .await
254− .unwrap();
255− }
256− Err(err) => {
257− error!("err: <- {:?}", err);
258− break;
259− }
260− }
261− }
262− }
263−
264− async fn send(
265− transport: Arc<Self>,
266− mut server_stdin: BufWriter<ChildStdin>,
267− mut client_rx: UnboundedReceiver<Request>,
268− ) {
269− while let Some(req) = client_rx.recv().await {
270− transport
271− .send_payload_to_server(&mut server_stdin, req)
272− .await
273− .unwrap()
274− }
275− }
276−}
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+}