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