git.iko.soy/helix
on 2158366 help sign in
helix/helix-dap/src/client.rs
398 lines 12 kB history raw
1use crate::{
2 transport::{Payload, Request, Transport},
3 types::*,
4 Error, Result,
5};
6use anyhow::anyhow;
7pub use log::{error, info};
8use std::{
9 collections::HashMap,
10 net::{IpAddr, Ipv4Addr, SocketAddr},
11 path::PathBuf,
12 process::Stdio,
13 sync::atomic::{AtomicU64, Ordering},
14};
15use tokio::{
16 io::{AsyncBufRead, AsyncWrite, BufReader, BufWriter},
17 net::TcpStream,
18 process::{Child, Command},
19 sync::mpsc::{channel, unbounded_channel, UnboundedReceiver, UnboundedSender},
20 time,
21};
23#[derive(Debug)]
24pub struct Client {
25 id: usize,
26 _process: Option<Child>,
27 server_tx: UnboundedSender<Request>,
28 request_counter: AtomicU64,
29 pub caps: Option<DebuggerCapabilities>,
30 //
31 pub breakpoints: HashMap<PathBuf, Vec<SourceBreakpoint>>,
32 // TODO: multiple threads support
33 pub stack_pointer: Option<StackFrame>,
34 pub stopped_thread: Option<usize>,
35 pub is_running: bool,
38impl Client {
39 // Spawn a process and communicate with it by either TCP or stdio
40 pub async fn process(
41 cfg: DebugAdapterConfig,
42 id: usize,
43 ) -> Result<(Self, UnboundedReceiver<Payload>)> {
44 if cfg.transport == "tcp" && cfg.port_arg.is_some() {
45 Self::tcp_process(
46 &cfg.command,
47 cfg.args.iter().map(|s| s.as_str()).collect(),
48 &cfg.port_arg.unwrap(),
49 id,
50 )
51 .await
52 } else if cfg.transport == "stdio" {
53 Self::stdio(
54 &cfg.command,
55 cfg.args.iter().map(|s| s.as_str()).collect(),
56 id,
57 )
58 } else {
59 Result::Err(Error::Other(anyhow!(
60 "Incorrect transport {}",
61 cfg.transport
62 )))
63 }
64 }
66 pub fn streams(
67 rx: Box<dyn AsyncBufRead + Unpin + Send>,
68 tx: Box<dyn AsyncWrite + Unpin + Send>,
69 id: usize,
70 process: Option<Child>,
71 ) -> Result<(Self, UnboundedReceiver<Payload>)> {
72 let (server_rx, server_tx) = Transport::start(rx, tx, id);
73 let (client_rx, client_tx) = unbounded_channel();
75 let client = Self {
76 id,
77 _process: process,
78 server_tx,
79 request_counter: AtomicU64::new(0),
80 caps: None,
81 //
82 breakpoints: HashMap::new(),
83 stack_pointer: None,
84 stopped_thread: None,
85 is_running: false,
86 };
88 tokio::spawn(Self::recv(server_rx, client_rx));
90 Ok((client, client_tx))
91 }
93 pub async fn tcp(
94 addr: std::net::SocketAddr,
95 id: usize,
96 ) -> Result<(Self, UnboundedReceiver<Payload>)> {
97 let stream = TcpStream::connect(addr).await?;
98 let (rx, tx) = stream.into_split();
99 Self::streams(Box::new(BufReader::new(rx)), Box::new(tx), id, None)
100 }
102 pub fn stdio(
103 cmd: &str,
104 args: Vec<&str>,
105 id: usize,
106 ) -> Result<(Self, UnboundedReceiver<Payload>)> {
107 let process = Command::new(cmd)
108 .args(args)
109 .stdin(Stdio::piped())
110 .stdout(Stdio::piped())
111 // make sure the process is reaped on drop
112 .kill_on_drop(true)
113 .spawn();
115 let mut process = process?;
117 // TODO: do we need bufreader/writer here? or do we use async wrappers on unblock?
118 let writer = BufWriter::new(process.stdin.take().expect("Failed to open stdin"));
119 let reader = BufReader::new(process.stdout.take().expect("Failed to open stdout"));
121 Self::streams(
122 Box::new(BufReader::new(reader)),
123 Box::new(writer),
124 id,
125 Some(process),
126 )
127 }
129 async fn get_port() -> Option<u16> {
130 Some(
131 tokio::net::TcpListener::bind(SocketAddr::new(
132 IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)),
133 0,
134 ))
135 .await
136 .ok()?
137 .local_addr()
138 .ok()?
139 .port(),
140 )
141 }
143 pub async fn tcp_process(
144 cmd: &str,
145 args: Vec<&str>,
146 port_format: &str,
147 id: usize,
148 ) -> Result<(Self, UnboundedReceiver<Payload>)> {
149 let port = Self::get_port().await.unwrap();
151 let process = Command::new(cmd)
152 .args(args)
153 .args(port_format.replace("{}", &port.to_string()).split(' '))
154 // silence messages
155 .stdin(Stdio::null())
156 .stdout(Stdio::null())
157 .stderr(Stdio::null())
158 // Do not kill debug adapter when leaving, it should exit automatically
159 .spawn()?;
161 // Wait for adapter to become ready for connection
162 time::sleep(time::Duration::from_millis(500)).await;
164 let stream = TcpStream::connect(SocketAddr::new(
165 IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)),
166 port,
167 ))
168 .await?;
170 let (rx, tx) = stream.into_split();
171 Self::streams(
172 Box::new(BufReader::new(rx)),
173 Box::new(tx),
174 id,
175 Some(process),
176 )
177 }
179 async fn recv(mut server_rx: UnboundedReceiver<Payload>, client_tx: UnboundedSender<Payload>) {
180 while let Some(msg) = server_rx.recv().await {
181 match msg {
182 Payload::Event(ev) => {
183 client_tx.send(Payload::Event(ev)).expect("Failed to send");
184 }
185 Payload::Response(_) => unreachable!(),
186 Payload::Request(req) => {
187 client_tx
188 .send(Payload::Request(req))
189 .expect("Failed to send");
190 }
191 }
192 }
193 }
195 pub fn id(&self) -> usize {
196 self.id
197 }
199 fn next_request_id(&self) -> u64 {
200 self.request_counter.fetch_add(1, Ordering::Relaxed)
201 }
203 async fn request<R: crate::types::Request>(
204 &self,
205 arguments: R::Arguments,
206 ) -> Result<R::Result> {
207 let (callback_tx, mut callback_rx) = channel(1);
209 let arguments = Some(serde_json::to_value(arguments)?);
211 let req = Request {
212 back_ch: Some(callback_tx),
213 seq: self.next_request_id(),
214 command: R::COMMAND.to_string(),
215 arguments,
216 };
218 self.server_tx
219 .send(req)
220 .expect("Failed to send request to debugger");
222 let response = callback_rx.recv().await.unwrap()?;
223 let response = serde_json::from_value(response.body.unwrap_or_default())?;
224 Ok(response)
225 }
227 pub fn capabilities(&self) -> &DebuggerCapabilities {
228 self.caps
229 .as_ref()
230 .expect("debugger not yet initialized!")
231 }
233 pub async fn initialize(&mut self, adapter_id: String) -> Result<()> {
234 let args = requests::InitializeArguments {
235 client_id: Some("hx".to_owned()),
236 client_name: Some("helix".to_owned()),
237 adapter_id,
238 locale: Some("en-us".to_owned()),
239 lines_start_at_one: Some(true),
240 columns_start_at_one: Some(true),
241 path_format: Some("path".to_owned()),
242 supports_variable_type: Some(true),
243 supports_variable_paging: Some(false),
244 supports_run_in_terminal_request: Some(false),
245 supports_memory_references: Some(false),
246 supports_progress_reporting: Some(false),
247 supports_invalidated_event: Some(false),
248 };
250 let response = self.request::<requests::Initialize>(args).await?;
251 self.caps = Some(response);
253 Ok(())
254 }
256 pub async fn disconnect(&mut self) -> Result<()> {
257 self.request::<requests::Disconnect>(()).await
258 }
260 pub async fn launch(&mut self, args: serde_json::Value) -> Result<()> {
261 let response = self.request::<requests::Launch>(args).await?;
262 log::error!("launch response {}", response);
264 Ok(())
265 }
267 pub async fn attach(&mut self, args: serde_json::Value) -> Result<()> {
268 let response = self.request::<requests::Attach>(args).await?;
269 log::error!("attach response {}", response);
271 Ok(())
272 }
274 pub async fn set_breakpoints(
275 &mut self,
276 file: PathBuf,
277 breakpoints: Vec<SourceBreakpoint>,
278 ) -> Result<Option<Vec<Breakpoint>>> {
279 let args = requests::SetBreakpointsArguments {
280 source: Source {
281 path: Some(file),
282 name: None,
283 source_reference: None,
284 presentation_hint: None,
285 origin: None,
286 sources: None,
287 adapter_data: None,
288 checksums: None,
289 },
290 breakpoints: Some(breakpoints),
291 source_modified: Some(false),
292 };
294 let response = self.request::<requests::SetBreakpoints>(args).await?;
296 Ok(response.breakpoints)
297 }
299 pub async fn configuration_done(&mut self) -> Result<()> {
300 self.request::<requests::ConfigurationDone>(()).await
301 }
303 pub async fn continue_thread(&mut self, thread_id: usize) -> Result<Option<bool>> {
304 let args = requests::ContinueArguments { thread_id };
306 let response = self.request::<requests::Continue>(args).await?;
307 Ok(response.all_threads_continued)
308 }
310 pub async fn stack_trace(
311 &mut self,
312 thread_id: usize,
313 ) -> Result<(Vec<StackFrame>, Option<usize>)> {
314 let args = requests::StackTraceArguments {
315 thread_id,
316 start_frame: None,
317 levels: None,
318 format: None,
319 };
321 let response = self.request::<requests::StackTrace>(args).await?;
322 Ok((response.stack_frames, response.total_frames))
323 }
325 pub async fn threads(&mut self) -> Result<Vec<Thread>> {
326 let response = self.request::<requests::Threads>(()).await?;
327 Ok(response.threads)
328 }
330 pub async fn scopes(&mut self, frame_id: usize) -> Result<Vec<Scope>> {
331 let args = requests::ScopesArguments { frame_id };
333 let response = self.request::<requests::Scopes>(args).await?;
334 Ok(response.scopes)
335 }
337 pub async fn variables(&mut self, variables_reference: usize) -> Result<Vec<Variable>> {
338 let args = requests::VariablesArguments {
339 variables_reference,
340 filter: None,
341 start: None,
342 count: None,
343 format: None,
344 };
346 let response = self.request::<requests::Variables>(args).await?;
347 Ok(response.variables)
348 }
350 pub async fn step_in(&mut self, thread_id: usize) -> Result<()> {
351 let args = requests::StepInArguments {
352 thread_id,
353 target_id: None,
354 granularity: None,
355 };
357 self.request::<requests::StepIn>(args).await
358 }
360 pub async fn step_out(&mut self, thread_id: usize) -> Result<()> {
361 let args = requests::StepOutArguments {
362 thread_id,
363 granularity: None,
364 };
366 self.request::<requests::StepOut>(args).await
367 }
369 pub async fn next(&mut self, thread_id: usize) -> Result<()> {
370 let args = requests::NextArguments {
371 thread_id,
372 granularity: None,
373 };
375 self.request::<requests::Next>(args).await
376 }
378 pub async fn pause(&mut self, thread_id: usize) -> Result<()> {
379 let args = requests::PauseArguments { thread_id };
381 self.request::<requests::Pause>(args).await
382 }
384 pub async fn eval(
385 &mut self,
386 expression: String,
387 frame_id: Option<usize>,
388 ) -> Result<requests::EvaluateResponse> {
389 let args = requests::EvaluateArguments {
390 expression,
391 frame_id,
392 context: None,
393 format: None,
394 };
396 self.request::<requests::Evaluate>(args).await
397 }