@@ -2,15 +2,22 @@ use crate::{
22 transport::{Event, Payload, Request, Response, Transport},
33 Result,
44};
5+use log::{error, info};
56use serde::{Deserialize, Serialize};
67use serde_json::{from_value, to_value, Value};
7−use std::process::Stdio;
8−use std::sync::atomic::{AtomicU64, Ordering};
8+use std::sync::{
9+ atomic::{AtomicU64, Ordering},
10+ Arc,
11+};
12+use std::{collections::HashMap, process::Stdio};
913use tokio::{
1014 io::{AsyncBufRead, AsyncWrite, BufReader, BufWriter},
1115 net::TcpStream,
1216 process::{Child, Command},
13− sync::mpsc::{channel, UnboundedReceiver, UnboundedSender},
17+ sync::{
18+ mpsc::{channel, Receiver, Sender, UnboundedReceiver, UnboundedSender},
19+ Mutex,
20+ },
1421};
1522
1623#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
@@ -252,9 +259,9 @@ pub struct Client {
252259 id: usize,
253260 _process: Option<Child>,
254261 server_tx: UnboundedSender<Request>,
255− server_rx: UnboundedReceiver<Payload>,
256262 request_counter: AtomicU64,
257263 capabilities: Option<DebuggerCapabilities>,
264+ awaited_events: Arc<Mutex<HashMap<String, Sender<Event>>>>,
258265}
259266
260267impl Client {
@@ -270,14 +277,52 @@ impl Client {
270277 id,
271278 _process: process,
272279 server_tx,
273− server_rx,
274280 request_counter: AtomicU64::new(0),
275281 capabilities: None,
282+ awaited_events: Arc::new(Mutex::new(HashMap::default())),
276283 };
277284
285+ tokio::spawn(Self::recv(Arc::clone(&client.awaited_events), server_rx));
286+
278287 Ok(client)
279288 }
280289
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+
281326 pub async fn tcp(addr: std::net::SocketAddr, id: usize) -> Result<Self> {
282327 let stream = TcpStream::connect(addr).await?;
283328 let (rx, tx) = stream.into_split();
@@ -373,45 +418,25 @@ impl Client {
373418 }
374419
375420 pub async fn launch(&mut self, args: impl Serialize) -> Result<()> {
421+ let mut initialized = self.listen_for_event("initialized".to_owned()).await;
422+
376423 self.request("launch".to_owned(), to_value(args).ok())
377424 .await?;
378425
379− match self
380− .server_rx
381− .recv()
382− .await
383− .expect("Expected initialized event")
384− {
385− Payload::Event(Event { event, .. }) => {
386− if event == *"initialized" {
387− Ok(())
388− } else {
389− unreachable!()
390− }
391− }
392− _ => unreachable!(),
393− }
426+ initialized.recv().await;
427+
428+ Ok(())
394429 }
395430
396431 pub async fn attach(&mut self, args: impl Serialize) -> Result<()> {
432+ let mut initialized = self.listen_for_event("initialized".to_owned()).await;
433+
397434 self.request("attach".to_owned(), to_value(args).ok())
398435 .await?;
399436
400− match self
401− .server_rx
402− .recv()
403− .await
404− .expect("Expected initialized event")
405− {
406− Payload::Event(Event { event, .. }) => {
407− if event == *"initialized" {
408− Ok(())
409− } else {
410− unreachable!()
411− }
412− }
413− _ => unreachable!(),
414− }
437+ initialized.recv().await;
438+
439+ Ok(())
415440 }
416441
417442 pub async fn set_breakpoints(
@@ -447,19 +472,6 @@ impl Client {
447472 Ok(())
448473 }
449474
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−
463475 pub async fn continue_thread(&mut self, thread_id: usize) -> Result<Option<bool>> {
464476 let args = ContinueArguments { thread_id };
465477