Implement RPC logging for debug adapter clients

Cargo.toml

@@ -172,6 +172,7 @@ members = [
     #
 
     "tooling/xtask",
+    "crates/debugger_tools",
 ]
 default-members = ["crates/zed"]
 
@@ -205,6 +206,7 @@ copilot = { path = "crates/copilot" }
 dap = { path = "crates/dap" }
 db = { path = "crates/db" }
 debugger_ui = { path = "crates/debugger_ui" }
+debugger_tools = { path = "crates/debugger_tools" }
 dev_server_projects = { path = "crates/dev_server_projects" }
 diagnostics = { path = "crates/diagnostics" }
 editor = { path = "crates/editor" }

crates/dap/src/client.rs

@@ -1,4 +1,5 @@
-use crate::transport::Transport;
+pub use crate::transport::IoKind;
+use crate::transport::{IoHandler, Transport};
 use anyhow::{anyhow, Context, Result};
 
 use crate::adapters::{build_adapter, DebugAdapter};
@@ -43,6 +44,7 @@ pub struct DebugAdapterClient {
     _process: Arc<Mutex<Option<Child>>>,
     sequence_count: AtomicU64,
     config: DebugAdapterConfig,
+    io_handlers: Arc<Mutex<Vec<IoHandler>>>,
 }
 
 pub struct TransportParams {
@@ -81,11 +83,14 @@ impl DebugAdapterClient {
         let adapter = Arc::new(build_adapter(&config).context("Creating debug adapter")?);
         let transport_params = adapter.connect(cx).await?;
 
+        let io_handlers = Arc::new(Mutex::new(Vec::new()));
+
         let transport = Self::handle_transport(
             transport_params.rx,
             transport_params.tx,
             transport_params.err,
             event_handler,
+            io_handlers.clone(),
             cx,
         );
 
@@ -94,6 +99,7 @@ impl DebugAdapterClient {
             config,
             adapter,
             transport,
+            io_handlers,
             sequence_count: AtomicU64::new(1),
             _process: Arc::new(Mutex::new(transport_params.process)),
         }))
@@ -104,12 +110,13 @@ impl DebugAdapterClient {
         tx: Box<dyn AsyncWrite + Unpin + Send>,
         err: Option<Box<dyn AsyncBufRead + Unpin + Send>>,
         event_handler: F,
+        io_handlers: Arc<Mutex<Vec<IoHandler>>>,
         cx: &mut AsyncAppContext,
     ) -> Arc<Transport>
     where
         F: FnMut(Message, &mut AppContext) + 'static + Send + Sync + Clone,
     {
-        let transport = Transport::start(rx, tx, err, cx);
+        let transport = Transport::start(rx, tx, err, io_handlers, cx);
 
         let server_rx = transport.server_rx.clone();
         let server_tr = transport.server_tx.clone();
@@ -231,4 +238,12 @@ impl DebugAdapterClient {
         }
         .await
     }
+
+    pub fn on_io<F>(&self, f: F)
+    where
+        F: 'static + Send + FnMut(IoKind, &str),
+    {
+        let mut io_handlers = self.io_handlers.lock();
+        io_handlers.push(Box::new(f));
+    }
 }

crates/dap/src/transport.rs

@@ -12,6 +12,15 @@ use smol::{
 };
 use std::{collections::HashMap, sync::Arc};
 
+pub type IoHandler = Box<dyn Send + FnMut(IoKind, &str)>;
+
+#[derive(Debug, Clone, Copy)]
+pub enum IoKind {
+    StdIn,
+    StdOut,
+    StdErr,
+}
+
 #[derive(Debug)]
 pub struct Transport {
     pub server_tx: Sender<Message>,
@@ -25,6 +34,7 @@ impl Transport {
         server_stdout: Box<dyn AsyncBufRead + Unpin + Send>,
         server_stdin: Box<dyn AsyncWrite + Unpin + Send>,
         server_stderr: Option<Box<dyn AsyncBufRead + Unpin + Send>>,
+        io_handlers: Arc<parking_lot::Mutex<Vec<IoHandler>>>,
         cx: &mut AsyncAppContext,
     ) -> Arc<Self> {
         let (client_tx, server_rx) = unbounded::<Message>();
@@ -38,11 +48,14 @@ impl Transport {
                 pending_requests.clone(),
                 server_stdout,
                 client_tx,
+                io_handlers.clone(),
             ))
             .detach();
 
         if let Some(stderr) = server_stderr {
-            cx.background_executor().spawn(Self::err(stderr)).detach();
+            cx.background_executor()
+                .spawn(Self::err(stderr, io_handlers.clone()))
+                .detach();
         }
 
         cx.background_executor()
@@ -51,6 +64,7 @@ impl Transport {
                 pending_requests.clone(),
                 server_stdin,
                 client_rx,
+                io_handlers,
             ))
             .detach();
 
@@ -65,6 +79,7 @@ impl Transport {
     async fn recv_server_message(
         reader: &mut Box<dyn AsyncBufRead + Unpin + Send>,
         buffer: &mut String,
+        io_handlers: Arc<parking_lot::Mutex<Vec<IoHandler>>>,
     ) -> Result<Message> {
         let mut content_length = None;
         loop {
@@ -102,18 +117,27 @@ impl Transport {
             .with_context(|| "reading after a loop")?;
 
         let msg = std::str::from_utf8(&content).context("invalid utf8 from server")?;
+
+        for handler in io_handlers.lock().iter_mut() {
+            handler(IoKind::StdOut, msg);
+        }
         Ok(serde_json::from_str::<Message>(msg)?)
     }
 
     async fn recv_server_error(
         err: &mut (impl AsyncBufRead + Unpin + Send),
         buffer: &mut String,
+        io_handlers: Arc<parking_lot::Mutex<Vec<IoHandler>>>,
     ) -> Result<()> {
         buffer.truncate(0);
         if err.read_line(buffer).await? == 0 {
             return Err(anyhow!("debugger error stream closed"));
         };
 
+        for handler in io_handlers.lock().iter_mut() {
+            handler(IoKind::StdErr, buffer.as_str());
+        }
+
         Ok(())
     }
 
@@ -122,6 +146,7 @@ impl Transport {
         pending_requests: &Mutex<HashMap<u64, Sender<Result<Response>>>>,
         server_stdin: &mut Box<dyn AsyncWrite + Unpin + Send>,
         mut payload: Message,
+        io_handlers: Arc<parking_lot::Mutex<Vec<IoHandler>>>,
     ) -> Result<()> {
         if let Message::Request(request) = &mut payload {
             {
@@ -130,17 +155,23 @@ impl Transport {
                 }
             }
         }
-        Self::send_string_to_server(server_stdin, serde_json::to_string(&payload)?).await
+        Self::send_string_to_server(server_stdin, serde_json::to_string(&payload)?, io_handlers)
+            .await
     }
 
     async fn send_string_to_server(
         server_stdin: &mut Box<dyn AsyncWrite + Unpin + Send>,
         request: String,
+        io_handlers: Arc<parking_lot::Mutex<Vec<IoHandler>>>,
     ) -> Result<()> {
         server_stdin
             .write_all(format!("Content-Length: {}\r\n\r\n{}", request.len(), request).as_bytes())
             .await?;
 
+        for handler in io_handlers.lock().iter_mut() {
+            handler(IoKind::StdIn, request.as_str());
+        }
+
         server_stdin.flush().await?;
         Ok(())
     }
@@ -188,10 +219,14 @@ impl Transport {
         pending_requests: Arc<Mutex<HashMap<u64, Sender<Result<Response>>>>>,
         mut server_stdout: Box<dyn AsyncBufRead + Unpin + Send>,
         client_tx: Sender<Message>,
+        io_handlers: Arc<parking_lot::Mutex<Vec<IoHandler>>>,
     ) -> Result<()> {
         let mut recv_buffer = String::new();
 
-        while let Ok(msg) = Self::recv_server_message(&mut server_stdout, &mut recv_buffer).await {
+        while let Ok(msg) =
+            Self::recv_server_message(&mut server_stdout, &mut recv_buffer, io_handlers.clone())
+                .await
+        {
             Self::process_server_message(&pending_requests, &client_tx, msg)
                 .await
                 .context("Process server message failed in transport::receive")?;
@@ -205,24 +240,30 @@ impl Transport {
         pending_requests: Arc<Mutex<HashMap<u64, Sender<Result<Response>>>>>,
         mut server_stdin: Box<dyn AsyncWrite + Unpin + Send>,
         client_rx: Receiver<Message>,
+        io_handlers: Arc<parking_lot::Mutex<Vec<IoHandler>>>,
     ) -> Result<()> {
         while let Ok(payload) = client_rx.recv().await {
             Self::send_payload_to_server(
                 &current_requests,
                 &pending_requests,
                 &mut server_stdin,
                 payload,
+                io_handlers.clone(),
             )
             .await?;
         }
 
         Ok(())
     }
 
-    async fn err(mut server_stderr: Box<dyn AsyncBufRead + Unpin + Send>) -> Result<()> {
+    async fn err(
+        mut server_stderr: Box<dyn AsyncBufRead + Unpin + Send>,
+        io_handlers: Arc<parking_lot::Mutex<Vec<IoHandler>>>,
+    ) -> Result<()> {
         let mut recv_buffer = String::new();
         loop {
-            Self::recv_server_error(&mut server_stderr, &mut recv_buffer).await?;
+            Self::recv_server_error(&mut server_stderr, &mut recv_buffer, io_handlers.clone())
+                .await?;
         }
     }
 }

crates/debugger_tools/Cargo.toml

@@ -0,0 +1,24 @@
+[package]
+name = "debugger_tools"
+version = "0.1.0"
+edition = "2021"
+publish = false
+license = "GPL-3.0-or-later"
+
+[lints]
+workspace = true
+
+[lib]
+path = "src/debugger_tools.rs"
+doctest = false
+
+[dependencies]
+gpui.workspace = true
+workspace.workspace = true
+editor.workspace = true
+project.workspace = true
+dap.workspace = true
+dap-types = { git = "https://github.com/zed-industries/dap-types" }
+serde_json.workspace = true
+futures.workspace = true
+anyhow.workspace = true

crates/debugger_tools/TODO.md

@@ -0,0 +1,4 @@
+- [ ] housekeeping: ensure references to debug adapters are cleaned when they are closed
+- [ ] feature: log server messages when available?
+- [ ] improve the client menu
+- [ ] log events separately?