git.iko.soy/helix
on master help sign in
dispatch events in client
d6de540Dmitry Sharshakov
helix-dap/examples/dap-basic.rs+5−1
compare with master
unifiedsplit
@@ -56,8 +56,12 @@ pub async fn main() -> Result<()> {
56 .read_line(&mut _in)
56 .read_line(&mut _in)
57 .expect("Failed to read line");
57 .expect("Failed to read line");
58
58
59+ let mut stopped_event = client.listen_for_event("stopped".to_owned()).await;
60+
59 println!("configurationDone: {:?}", client.configuration_done().await);
61 println!("configurationDone: {:?}", client.configuration_done().await);
62+
60− println!("stopped: {:?}", client.wait_for_stopped().await);
63+ println!("stopped: {:?}", stopped_event.recv().await);
64+
61 println!("threads: {:#?}", client.threads().await);
65 println!("threads: {:#?}", client.threads().await);
62 let bt = client.stack_trace(1).await.expect("expected stack trace");
66 let bt = client.stack_trace(1).await.expect("expected stack trace");
63 println!("stack trace: {:#?}", bt);
67 println!("stack trace: {:#?}", bt);
helix-dap/src/client.rs+60−48
@@ -2,15 +2,22 @@ use crate::{
2 transport::{Event, Payload, Request, Response, Transport},
2 transport::{Event, Payload, Request, Response, Transport},
3 Result,
3 Result,
4};
4};
5+use log::{error, info};
5use serde::{Deserialize, Serialize};
6use serde::{Deserialize, Serialize};
6use serde_json::{from_value, to_value, Value};
7use serde_json::{from_value, to_value, Value};
7−use std::process::Stdio;
8+use std::sync::{
8−use std::sync::atomic::{AtomicU64, Ordering};
9+ atomic::{AtomicU64, Ordering},
10+ Arc,
11+};
12+use std::{collections::HashMap, process::Stdio};
9use tokio::{
13use tokio::{
10 io::{AsyncBufRead, AsyncWrite, BufReader, BufWriter},
14 io::{AsyncBufRead, AsyncWrite, BufReader, BufWriter},
11 net::TcpStream,
15 net::TcpStream,
12 process::{Child, Command},
16 process::{Child, Command},
17+ sync::{
13− sync::mpsc::{channel, UnboundedReceiver, UnboundedSender},
18+ mpsc::{channel, Receiver, Sender, UnboundedReceiver, UnboundedSender},
19+ Mutex,
20+ },
14};
21};
15
22
16#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
23#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
@@ -252,9 +259,9 @@ pub struct Client {
252 id: usize,
259 id: usize,
253 _process: Option<Child>,
260 _process: Option<Child>,
254 server_tx: UnboundedSender<Request>,
261 server_tx: UnboundedSender<Request>,
255− server_rx: UnboundedReceiver<Payload>,
256 request_counter: AtomicU64,
262 request_counter: AtomicU64,
257 capabilities: Option<DebuggerCapabilities>,
263 capabilities: Option<DebuggerCapabilities>,
264+ awaited_events: Arc<Mutex<HashMap<String, Sender<Event>>>>,
258}
265}
259
266
260impl Client {
267impl Client {
@@ -270,14 +277,52 @@ impl Client {
270 id,
277 id,
271 _process: process,
278 _process: process,
272 server_tx,
279 server_tx,
273− server_rx,
274 request_counter: AtomicU64::new(0),
280 request_counter: AtomicU64::new(0),
275 capabilities: None,
281 capabilities: None,
282+ awaited_events: Arc::new(Mutex::new(HashMap::default())),
276 };
283 };
277
284
285+ tokio::spawn(Self::recv(Arc::clone(&client.awaited_events), server_rx));
286+
278 Ok(client)
287 Ok(client)
279 }
288 }
280
289
290+ async fn recv(
291+ awaited_events: Arc<Mutex<HashMap<String, Sender<Event>>>>,
292+ mut server_rx: UnboundedReceiver<Payload>,
293+ ) {
294+ while let Some(msg) = server_rx.recv().await {
295+ match msg {
296+ Payload::Event(ev) => {
297+ let name = ev.event.clone();
298+ let tx = awaited_events.lock().await.remove(&name);
299+
300+ match tx {
301+ Some(tx) => match tx.send(ev).await {
302+ Ok(_) => (),
303+ Err(_) => error!(
304+ "Tried sending event into a closed channel (name={:?})",
305+ name
306+ ),
307+ },
308+ None => {
309+ info!("unhandled event");
310+ // client_tx.send(Payload::Event(ev)).expect("Failed to send");
311+ }
312+ }
313+ }
314+ Payload::Response(_) => unreachable!(),
315+ Payload::Request(_) => todo!(),
316+ }
317+ }
318+ }
319+
320+ pub async fn listen_for_event(&self, name: String) -> Receiver<Event> {
321+ let (rx, tx) = channel(1);
322+ self.awaited_events.lock().await.insert(name.clone(), rx);
323+ tx
324+ }
325+
281 pub async fn tcp(addr: std::net::SocketAddr, id: usize) -> Result<Self> {
326 pub async fn tcp(addr: std::net::SocketAddr, id: usize) -> Result<Self> {
282 let stream = TcpStream::connect(addr).await?;
327 let stream = TcpStream::connect(addr).await?;
283 let (rx, tx) = stream.into_split();
328 let (rx, tx) = stream.into_split();
@@ -373,45 +418,25 @@ impl Client {
373 }
418 }
374
419
375 pub async fn launch(&mut self, args: impl Serialize) -> Result<()> {
420 pub async fn launch(&mut self, args: impl Serialize) -> Result<()> {
421+ let mut initialized = self.listen_for_event("initialized".to_owned()).await;
422+
376 self.request("launch".to_owned(), to_value(args).ok())
423 self.request("launch".to_owned(), to_value(args).ok())
377 .await?;
424 .await?;
378
425
379− match self
380− .server_rx
381− .recv()
382− .await
426+ initialized.recv().await;
383− .expect("Expected initialized event")
427+
384− {
385− Payload::Event(Event { event, .. }) => {
386− if event == *"initialized" {
387− Ok(())
428+ Ok(())
388− } else {
389− unreachable!()
390− }
391− }
392− _ => unreachable!(),
393− }
394 }
429 }
395
430
396 pub async fn attach(&mut self, args: impl Serialize) -> Result<()> {
431 pub async fn attach(&mut self, args: impl Serialize) -> Result<()> {
432+ let mut initialized = self.listen_for_event("initialized".to_owned()).await;
433+
397 self.request("attach".to_owned(), to_value(args).ok())
434 self.request("attach".to_owned(), to_value(args).ok())
398 .await?;
435 .await?;
399
436
400− match self
401− .server_rx
402− .recv()
403− .await
437+ initialized.recv().await;
404− .expect("Expected initialized event")
438+
405− {
406− Payload::Event(Event { event, .. }) => {
407− if event == *"initialized" {
408− Ok(())
439+ Ok(())
409− } else {
410− unreachable!()
411− }
412− }
413− _ => unreachable!(),
414− }
415 }
440 }
416
441
417 pub async fn set_breakpoints(
442 pub async fn set_breakpoints(
@@ -447,19 +472,6 @@ impl Client {
447 Ok(())
472 Ok(())
448 }
473 }
449
474
450− pub async fn wait_for_stopped(&mut self) -> Result<()> {
451− match self.server_rx.recv().await.expect("Expected stopped event") {
452− Payload::Event(Event { event, .. }) => {
453− if event == *"stopped" {
454− Ok(())
455− } else {
456− unreachable!()
457− }
458− }
459− _ => unreachable!(),
460− }
461− }
462−
463 pub async fn continue_thread(&mut self, thread_id: usize) -> Result<Option<bool>> {
475 pub async fn continue_thread(&mut self, thread_id: usize) -> Result<Option<bool>> {
464 let args = ContinueArguments { thread_id };
476 let args = ContinueArguments { thread_id };
465
477