diff options
| -rw-r--r-- | Cargo.lock | 1 | ||||
| -rw-r--r-- | helix-dap/src/client.rs | 74 | ||||
| -rw-r--r-- | helix-term/Cargo.toml | 2 | ||||
| -rw-r--r-- | helix-term/src/application.rs | 24 | ||||
| -rw-r--r-- | helix-term/src/commands.rs | 36 | ||||
| -rw-r--r-- | helix-term/src/ui/editor.rs | 19 | ||||
| -rw-r--r-- | helix-view/src/editor.rs | 7 |
7 files changed, 65 insertions, 98 deletions
@@ -389,6 +389,7 @@ dependencies = [ "signal-hook", "signal-hook-tokio", "tokio", + "tokio-stream", "toml", ] diff --git a/helix-dap/src/client.rs b/helix-dap/src/client.rs index f2d5b865..bc7a93d8 100644 --- a/helix-dap/src/client.rs +++ b/helix-dap/src/client.rs @@ -1,5 +1,5 @@ use crate::{ - transport::{Event, Payload, Request, Transport}, + transport::{Payload, Request, Transport}, types::*, Result, }; @@ -9,19 +9,13 @@ use std::{ net::{IpAddr, Ipv4Addr, SocketAddr}, path::PathBuf, process::Stdio, - sync::{ - atomic::{AtomicU64, Ordering}, - Arc, - }, + sync::atomic::{AtomicU64, Ordering}, }; use tokio::{ io::{AsyncBufRead, AsyncWrite, BufReader, BufWriter}, net::TcpStream, process::{Child, Command}, - sync::{ - mpsc::{channel, Receiver, Sender, UnboundedReceiver, UnboundedSender}, - Mutex, - }, + sync::mpsc::{channel, unbounded_channel, UnboundedReceiver, UnboundedSender}, time, }; @@ -32,8 +26,6 @@ pub struct Client { server_tx: UnboundedSender<Request>, request_counter: AtomicU64, capabilities: Option<DebuggerCapabilities>, - awaited_events: Arc<Mutex<HashMap<String, Sender<Event>>>>, - // pub breakpoints: HashMap<PathBuf, Vec<SourceBreakpoint>>, // TODO: multiple threads support @@ -46,8 +38,9 @@ impl Client { tx: Box<dyn AsyncWrite + Unpin + Send>, id: usize, process: Option<Child>, - ) -> Result<Self> { + ) -> Result<(Self, UnboundedReceiver<Payload>)> { let (server_rx, server_tx) = Transport::start(rx, tx, id); + let (client_rx, client_tx) = unbounded_channel(); let client = Self { id, @@ -55,24 +48,30 @@ impl Client { server_tx, request_counter: AtomicU64::new(0), capabilities: None, - awaited_events: Arc::new(Mutex::new(HashMap::default())), // breakpoints: HashMap::new(), stack_pointer: None, }; - tokio::spawn(Self::recv(Arc::clone(&client.awaited_events), server_rx)); + tokio::spawn(Self::recv(server_rx, client_rx)); - Ok(client) + Ok((client, client_tx)) } - pub async fn tcp(addr: std::net::SocketAddr, id: usize) -> Result<Self> { + pub async fn tcp( + addr: std::net::SocketAddr, + id: usize, + ) -> Result<(Self, UnboundedReceiver<Payload>)> { let stream = TcpStream::connect(addr).await?; let (rx, tx) = stream.into_split(); Self::streams(Box::new(BufReader::new(rx)), Box::new(tx), id, None) } - pub fn stdio(cmd: &str, args: Vec<&str>, id: usize) -> Result<Self> { + pub fn stdio( + cmd: &str, + args: Vec<&str>, + id: usize, + ) -> Result<(Self, UnboundedReceiver<Payload>)> { let process = Command::new(cmd) .args(args) .stdin(Stdio::piped()) @@ -114,7 +113,7 @@ impl Client { args: Vec<&str>, port_format: &str, id: usize, - ) -> Result<Self> { + ) -> Result<(Self, UnboundedReceiver<Payload>)> { let port = Self::get_port().await.unwrap(); let process = Command::new(cmd) @@ -145,43 +144,22 @@ impl Client { ) } - async fn recv( - awaited_events: Arc<Mutex<HashMap<String, Sender<Event>>>>, - mut server_rx: UnboundedReceiver<Payload>, - ) { + async fn recv(mut server_rx: UnboundedReceiver<Payload>, client_tx: UnboundedSender<Payload>) { while let Some(msg) = server_rx.recv().await { match msg { Payload::Event(ev) => { - let name = ev.event.clone(); - let hashmap = awaited_events.lock().await; - let tx = hashmap.get(&name); - - match tx { - Some(tx) => match tx.send(ev).await { - Ok(_) => (), - Err(_) => error!( - "Tried sending event into a closed channel (name={:?})", - name - ), - }, - None => { - info!("unhandled event {}", name); - // client_tx.send(Payload::Event(ev)).expect("Failed to send"); - } - } + client_tx.send(Payload::Event(ev)).expect("Failed to send"); } Payload::Response(_) => unreachable!(), - Payload::Request(_) => todo!(), + Payload::Request(req) => { + client_tx + .send(Payload::Request(req)) + .expect("Failed to send"); + } } } } - pub async fn listen_for_event(&self, name: String) -> Receiver<Event> { - let (rx, tx) = channel(1); - self.awaited_events.lock().await.insert(name.clone(), rx); - tx - } - pub fn id(&self) -> usize { self.id } @@ -248,8 +226,6 @@ impl Client { } pub async fn launch(&mut self, args: serde_json::Value) -> Result<()> { - // TODO: buffer these until initialized arrives - let response = self.request::<requests::Launch>(args).await?; log::error!("launch response {}", response); @@ -257,8 +233,6 @@ impl Client { } pub async fn attach(&mut self, args: serde_json::Value) -> Result<()> { - // TODO: buffer these until initialized arrives - let response = self.request::<requests::Attach>(args).await?; log::error!("attach response {}", response); diff --git a/helix-term/Cargo.toml b/helix-term/Cargo.toml index 6ed60e02..c0b2cfe5 100644 --- a/helix-term/Cargo.toml +++ b/helix-term/Cargo.toml @@ -34,7 +34,7 @@ num_cpus = "1" tui = { path = "../helix-tui", package = "helix-tui", default-features = false, features = ["crossterm"] } crossterm = { version = "0.20", features = ["event-stream"] } signal-hook = "0.3" - +tokio-stream = "0.1" futures-util = { version = "0.3", features = ["std", "async-await"], default-features = false } # Logging diff --git a/helix-term/src/application.rs b/helix-term/src/application.rs index 011dafee..dad6df2b 100644 --- a/helix-term/src/application.rs +++ b/helix-term/src/application.rs @@ -1,4 +1,5 @@ use helix_core::syntax; +use helix_dap::Payload; use helix_lsp::{lsp, util::lsp_pos_to_pos, LspProgressMap}; use helix_view::{theme, Editor}; @@ -184,6 +185,29 @@ impl Application { last_render = Instant::now(); } } + Some(payload) = self.editor.debugger_events.next() => { + let mut debugger = self.editor.debugger.as_mut().unwrap(); + match payload { + Payload::Event(ev) => { + match &ev.event[..] { + "stopped" => { + let main = debugger + .threads() + .await + .ok() + .and_then(|threads| threads.get(0).cloned()); + if let Some(main) = main { + let (bt, _) = debugger.stack_trace(main.id).await.unwrap(); + debugger.stack_pointer = bt.get(0).cloned(); + } + } + _ => {} + } + }, + Payload::Response(_) => unreachable!(), + Payload::Request(_) => todo!(), + } + } Some(callback) = self.jobs.futures.next() => { self.jobs.handle_callback(&mut self.editor, &mut self.compositor, callback); self.render(); diff --git a/helix-term/src/commands.rs b/helix-term/src/commands.rs index 9a8e1ee6..34a146c2 100644 --- a/helix-term/src/commands.rs +++ b/helix-term/src/commands.rs @@ -24,17 +24,17 @@ use helix_lsp::{ }; use insert::*; use movement::Movement; -use tokio::sync::{mpsc::Receiver, Mutex}; use crate::{ compositor::{self, Component, Compositor}, ui::{self, FilePicker, Picker, Popup, Prompt, PromptEvent}, }; +use tokio_stream::wrappers::UnboundedReceiverStream; use crate::job::{self, Job, Jobs}; use futures_util::FutureExt; +use std::num::NonZeroUsize; use std::{collections::HashMap, fmt, future::Future}; -use std::{num::NonZeroUsize, sync::Arc}; use std::{ borrow::Cow, @@ -4255,8 +4255,8 @@ fn dap_start(cx: &mut Context) { // look up config for filetype // if multiple available, open picker - let debugger = Client::tcp_process("dlv", vec!["dap"], "-l 127.0.0.1:{}", 0); - let mut debugger = block_on(debugger).unwrap(); + let started = Client::tcp_process("dlv", vec!["dap"], "-l 127.0.0.1:{}", 0); + let (mut debugger, events) = block_on(started).unwrap(); let request = debugger.initialize("go".to_owned()); let _ = block_on(request).unwrap(); @@ -4268,32 +4268,10 @@ fn dap_start(cx: &mut Context) { let request = debugger.launch(to_value(args).unwrap()); let _ = block_on(request).unwrap(); - let stopped = block_on(debugger.listen_for_event("stopped".to_owned())); - let debugger = Arc::new(Mutex::new(debugger)); - tokio::spawn(dap_listen_stopped(stopped, Arc::clone(&debugger))); - // TODO: either await "initialized" or buffer commands until event is received cx.editor.debugger = Some(debugger); -} - -async fn dap_listen_stopped( - mut stopped: Receiver<helix_dap::Event>, - debugger: Arc<Mutex<helix_dap::Client>>, -) { - loop { - stopped.recv().await; - - let mut dbg = debugger.lock().await; - let main = dbg - .threads() - .await - .ok() - .and_then(|threads| threads.get(0).cloned()); - if let Some(main) = main { - let (a, _) = dbg.stack_trace(main.id).await.unwrap(); - dbg.stack_pointer = a.get(0).cloned(); - } - } + let stream = UnboundedReceiverStream::new(events); + cx.editor.debugger_events.push(stream); } fn dap_toggle_breakpoint(cx: &mut Context) { @@ -4321,7 +4299,6 @@ fn dap_toggle_breakpoint(cx: &mut Context) { // we shouldn't really allow editing while debug is running though if let Some(debugger) = &mut cx.editor.debugger { - let mut debugger = block_on(debugger.lock()); let breakpoints = debugger.breakpoints.entry(path.clone()).or_default(); if let Some(pos) = breakpoints.iter().position(|b| b.line == breakpoint.line) { breakpoints.remove(pos); @@ -4340,7 +4317,6 @@ fn dap_run(cx: &mut Context) { use helix_lsp::block_on; if let Some(debugger) = &mut cx.editor.debugger { - let mut debugger = block_on(debugger.lock()); let request = debugger.configuration_done(); let _ = block_on(request).unwrap(); } diff --git a/helix-term/src/ui/editor.rs b/helix-term/src/ui/editor.rs index 7f441a7d..f8035ae4 100644 --- a/helix-term/src/ui/editor.rs +++ b/helix-term/src/ui/editor.rs @@ -25,8 +25,7 @@ use helix_view::{ keyboard::{KeyCode, KeyModifiers}, Document, Editor, Theme, View, }; -use std::{borrow::Cow, sync::Arc}; -use tokio::sync::Mutex; +use std::borrow::Cow; use crossterm::event::{Event, MouseButton, MouseEvent, MouseEventKind}; use tui::buffer::Buffer as Surface; @@ -73,7 +72,7 @@ impl EditorView { is_focused: bool, loader: &syntax::Loader, config: &helix_view::editor::Config, - debugger: Option<Arc<Mutex<helix_dap::Client>>>, + debugger: &Option<helix_dap::Client>, ) { let inner = view.inner_area(); let area = view.area; @@ -414,9 +413,8 @@ impl EditorView { theme: &Theme, is_focused: bool, config: &helix_view::editor::Config, - debugger: Option<Arc<Mutex<helix_dap::Client>>>, + debugger: &Option<helix_dap::Client>, ) { - use helix_lsp::block_on; let text = doc.text().slice(..); let last_line = view.last_line(doc); @@ -449,9 +447,8 @@ impl EditorView { let mut stack_pointer: Option<StackFrame> = None; if let Some(debugger) = debugger { if let Some(path) = doc.path() { - let dbg = block_on(debugger.lock()); - breakpoints = dbg.breakpoints.get(path).cloned(); - stack_pointer = dbg.stack_pointer.clone() + breakpoints = debugger.breakpoints.get(path).cloned(); + stack_pointer = debugger.stack_pointer.clone() } } @@ -1047,10 +1044,6 @@ impl Component for EditorView { for (view, is_focused) in cx.editor.tree.views() { let doc = cx.editor.document(view.doc).unwrap(); let loader = &cx.editor.syn_loader; - let mut dbg: Option<Arc<Mutex<helix_dap::Client>>> = None; - if let Some(debugger) = &cx.editor.debugger { - dbg = Some(Arc::clone(debugger)); - } self.render_view( doc, view, @@ -1060,7 +1053,7 @@ impl Component for EditorView { is_focused, loader, &cx.editor.config, - dbg, + &cx.editor.debugger, ); } diff --git a/helix-view/src/editor.rs b/helix-view/src/editor.rs index 2309379a..17319327 100644 --- a/helix-view/src/editor.rs +++ b/helix-view/src/editor.rs @@ -8,7 +8,6 @@ use crate::{ use futures_util::future; use futures_util::stream::select_all::SelectAll; -use tokio::sync::Mutex; use tokio_stream::wrappers::UnboundedReceiverStream; use std::{ @@ -75,8 +74,8 @@ pub struct Editor { pub theme: Theme, pub language_servers: helix_lsp::Registry, - pub debugger: Option<Arc<Mutex<helix_dap::Client>>>, - pub debuggers: SelectAll<UnboundedReceiverStream<helix_dap::Payload>>, + pub debugger: Option<helix_dap::Client>, + pub debugger_events: SelectAll<UnboundedReceiverStream<helix_dap::Payload>>, pub clipboard_provider: Box<dyn ClipboardProvider>, @@ -116,7 +115,7 @@ impl Editor { theme: themes.default(), language_servers, debugger: None, - debuggers: SelectAll::new(), + debugger_events: SelectAll::new(), syn_loader: config_loader, theme_loader: themes, registers: Registers::default(), |