@@ -1,5 +1,5 @@
11use crate::{
2− transport::{Event, Payload, Request, Transport},
2+ transport::{Payload, Request, Transport},
33 types::*,
44 Result,
55};
@@ -9,19 +9,13 @@ use std::{
99 net::{IpAddr, Ipv4Addr, SocketAddr},
1010 path::PathBuf,
1111 process::Stdio,
12− sync::{
13− atomic::{AtomicU64, Ordering},
14− Arc,
15− },
12+ sync::atomic::{AtomicU64, Ordering},
1613};
1714use tokio::{
1815 io::{AsyncBufRead, AsyncWrite, BufReader, BufWriter},
1916 net::TcpStream,
2017 process::{Child, Command},
21− sync::{
22− mpsc::{channel, Receiver, Sender, UnboundedReceiver, UnboundedSender},
23− Mutex,
24− },
18+ sync::mpsc::{channel, unbounded_channel, UnboundedReceiver, UnboundedSender},
2519 time,
2620};
2721
@@ -32,8 +26,6 @@ pub struct Client {
3226 server_tx: UnboundedSender<Request>,
3327 request_counter: AtomicU64,
3428 capabilities: Option<DebuggerCapabilities>,
35− awaited_events: Arc<Mutex<HashMap<String, Sender<Event>>>>,
36−
3729 //
3830 pub breakpoints: HashMap<PathBuf, Vec<SourceBreakpoint>>,
3931 // TODO: multiple threads support
@@ -46,8 +38,9 @@ impl Client {
4638 tx: Box<dyn AsyncWrite + Unpin + Send>,
4739 id: usize,
4840 process: Option<Child>,
49− ) -> Result<Self> {
41+ ) -> Result<(Self, UnboundedReceiver<Payload>)> {
5042 let (server_rx, server_tx) = Transport::start(rx, tx, id);
43+ let (client_rx, client_tx) = unbounded_channel();
5144
5245 let client = Self {
5346 id,
@@ -55,24 +48,30 @@ impl Client {
5548 server_tx,
5649 request_counter: AtomicU64::new(0),
5750 capabilities: None,
58− awaited_events: Arc::new(Mutex::new(HashMap::default())),
5951 //
6052 breakpoints: HashMap::new(),
6153 stack_pointer: None,
6254 };
6355
64− tokio::spawn(Self::recv(Arc::clone(&client.awaited_events), server_rx));
56+ tokio::spawn(Self::recv(server_rx, client_rx));
6557
66− Ok(client)
58+ Ok((client, client_tx))
6759 }
6860
69− pub async fn tcp(addr: std::net::SocketAddr, id: usize) -> Result<Self> {
61+ pub async fn tcp(
62+ addr: std::net::SocketAddr,
63+ id: usize,
64+ ) -> Result<(Self, UnboundedReceiver<Payload>)> {
7065 let stream = TcpStream::connect(addr).await?;
7166 let (rx, tx) = stream.into_split();
7267 Self::streams(Box::new(BufReader::new(rx)), Box::new(tx), id, None)
7368 }
7469
75− pub fn stdio(cmd: &str, args: Vec<&str>, id: usize) -> Result<Self> {
70+ pub fn stdio(
71+ cmd: &str,
72+ args: Vec<&str>,
73+ id: usize,
74+ ) -> Result<(Self, UnboundedReceiver<Payload>)> {
7675 let process = Command::new(cmd)
7776 .args(args)
7877 .stdin(Stdio::piped())
@@ -114,7 +113,7 @@ impl Client {
114113 args: Vec<&str>,
115114 port_format: &str,
116115 id: usize,
117− ) -> Result<Self> {
116+ ) -> Result<(Self, UnboundedReceiver<Payload>)> {
118117 let port = Self::get_port().await.unwrap();
119118
120119 let process = Command::new(cmd)
@@ -145,43 +144,22 @@ impl Client {
145144 )
146145 }
147146
148− async fn recv(
149− awaited_events: Arc<Mutex<HashMap<String, Sender<Event>>>>,
150− mut server_rx: UnboundedReceiver<Payload>,
151− ) {
147+ async fn recv(mut server_rx: UnboundedReceiver<Payload>, client_tx: UnboundedSender<Payload>) {
152148 while let Some(msg) = server_rx.recv().await {
153149 match msg {
154150 Payload::Event(ev) => {
155− let name = ev.event.clone();
156− let hashmap = awaited_events.lock().await;
157− let tx = hashmap.get(&name);
158−
159− match tx {
160− Some(tx) => match tx.send(ev).await {
161− Ok(_) => (),
162− Err(_) => error!(
163− "Tried sending event into a closed channel (name={:?})",
164− name
165− ),
166− },
167− None => {
168− info!("unhandled event {}", name);
169− // client_tx.send(Payload::Event(ev)).expect("Failed to send");
170− }
171− }
151+ client_tx.send(Payload::Event(ev)).expect("Failed to send");
172152 }
173153 Payload::Response(_) => unreachable!(),
174− Payload::Request(_) => todo!(),
154+ Payload::Request(req) => {
155+ client_tx
156+ .send(Payload::Request(req))
157+ .expect("Failed to send");
158+ }
175159 }
176160 }
177161 }
178162
179− pub async fn listen_for_event(&self, name: String) -> Receiver<Event> {
180− let (rx, tx) = channel(1);
181− self.awaited_events.lock().await.insert(name.clone(), rx);
182− tx
183− }
184−
185163 pub fn id(&self) -> usize {
186164 self.id
187165 }
@@ -248,8 +226,6 @@ impl Client {
248226 }
249227
250228 pub async fn launch(&mut self, args: serde_json::Value) -> Result<()> {
251− // TODO: buffer these until initialized arrives
252−
253229 let response = self.request::<requests::Launch>(args).await?;
254230 log::error!("launch response {}", response);
255231
@@ -257,8 +233,6 @@ impl Client {
257233 }
258234
259235 pub async fn attach(&mut self, args: serde_json::Value) -> Result<()> {
260− // TODO: buffer these until initialized arrives
261−
262236 let response = self.request::<requests::Attach>(args).await?;
263237 log::error!("attach response {}", response);
264238