Abstract over channel

crates/ra_flycheck/src/lib.rs

@@ -52,17 +52,19 @@ pub struct FlycheckHandle {
     // XXX: drop order is significant
     cmd_send: Sender<CheckCommand>,
     handle: jod_thread::JoinHandle<()>,
-    pub task_recv: Receiver<CheckTask>,
 }
 
 impl FlycheckHandle {
-    pub fn spawn(config: FlycheckConfig, workspace_root: PathBuf) -> FlycheckHandle {
+    pub fn spawn(
-        let (task_send, task_recv) = unbounded::<CheckTask>();
+        config: FlycheckConfig,
+        workspace_root: PathBuf,
+        sender: Box<dyn Fn(CheckTask) + Send>,
+    ) -> FlycheckHandle {
         let (cmd_send, cmd_recv) = unbounded::<CheckCommand>();
         let handle = jod_thread::spawn(move || {
-            FlycheckActor::new(config, workspace_root).run(&task_send, &cmd_recv);
+            FlycheckActor::new(config, workspace_root, sender).run(&cmd_recv);
         });
-        FlycheckHandle { task_recv, cmd_send, handle }
+        FlycheckHandle { cmd_send, handle }
     }
 
     /// Schedule a re-start of the cargo check worker.
@@ -96,6 +98,7 @@ pub enum CheckCommand {
 }
 
 struct FlycheckActor {
+    sender: Box<dyn Fn(CheckTask) + Send>,
     config: FlycheckConfig,
     workspace_root: PathBuf,
     last_update_req: Option<Instant>,
@@ -110,8 +113,13 @@ struct FlycheckActor {
 }
 
 impl FlycheckActor {
-    fn new(config: FlycheckConfig, workspace_root: PathBuf) -> FlycheckActor {
+    fn new(
+        config: FlycheckConfig,
+        workspace_root: PathBuf,
+        sender: Box<dyn Fn(CheckTask) + Send>,
+    ) -> FlycheckActor {
         FlycheckActor {
+            sender,
             config,
             workspace_root,
             last_update_req: None,
@@ -120,9 +128,9 @@ impl FlycheckActor {
         }
     }
 
-    fn run(&mut self, task_send: &Sender<CheckTask>, cmd_recv: &Receiver<CheckCommand>) {
+    fn run(&mut self, cmd_recv: &Receiver<CheckCommand>) {
         // If we rerun the thread, we need to discard the previous check results first
-        self.clean_previous_results(task_send);
+        self.clean_previous_results();
 
         loop {
             select! {
@@ -134,7 +142,7 @@ impl FlycheckActor {
                     },
                 },
                 recv(self.message_recv) -> msg => match msg {
-                    Ok(msg) => self.handle_message(msg, task_send),
+                    Ok(msg) => self.handle_message(msg),
                     Err(RecvError) => {
                         // Watcher finished, replace it with a never channel to
                         // avoid busy-waiting.
@@ -146,15 +154,15 @@ impl FlycheckActor {
 
             if self.should_recheck() {
                 self.last_update_req = None;
-                task_send.send(CheckTask::ClearDiagnostics).unwrap();
+                self.send(CheckTask::ClearDiagnostics);
                 self.restart_check_process();
             }
         }
     }
 
-    fn clean_previous_results(&self, task_send: &Sender<CheckTask>) {
+    fn clean_previous_results(&self) {
-        task_send.send(CheckTask::ClearDiagnostics).unwrap();
+        self.send(CheckTask::ClearDiagnostics);
-        task_send.send(CheckTask::Status(Status::End)).unwrap();
+        self.send(CheckTask::Status(Status::End));
     }
 
     fn should_recheck(&mut self) -> bool {
@@ -173,27 +181,25 @@ impl FlycheckActor {
         }
     }
 
-    fn handle_message(&self, msg: CheckEvent, task_send: &Sender<CheckTask>) {
+    fn handle_message(&self, msg: CheckEvent) {
         match msg {
             CheckEvent::Begin => {
-                task_send.send(CheckTask::Status(Status::Being)).unwrap();
+                self.send(CheckTask::Status(Status::Being));
             }
 
             CheckEvent::End => {
-                task_send.send(CheckTask::Status(Status::End)).unwrap();
+                self.send(CheckTask::Status(Status::End));
             }
 
             CheckEvent::Msg(Message::CompilerArtifact(msg)) => {
-                task_send.send(CheckTask::Status(Status::Progress(msg.target.name))).unwrap();
+                self.send(CheckTask::Status(Status::Progress(msg.target.name)));
             }
 
             CheckEvent::Msg(Message::CompilerMessage(msg)) => {
-                task_send
+                self.send(CheckTask::AddDiagnostic {
-                    .send(CheckTask::AddDiagnostic {
                     workspace_root: self.workspace_root.clone(),
                     diagnostic: msg.message,
-                    })
+                });
-                    .unwrap();
             }
 
             CheckEvent::Msg(Message::BuildScriptExecuted(_msg)) => {}
@@ -271,6 +277,10 @@ impl FlycheckActor {
             let _ = message_send.send(CheckEvent::End);
         }))
     }
+
+    fn send(&self, check_task: CheckTask) {
+        (self.sender)(check_task)
+    }
 }
 
 enum CheckEvent {

crates/rust-analyzer/src/global_state.rs

@@ -9,7 +9,7 @@ use crossbeam_channel::{unbounded, Receiver};
 use lsp_types::Url;
 use parking_lot::RwLock;
 use ra_db::{CrateId, SourceRoot, VfsPath};
-use ra_flycheck::{FlycheckConfig, FlycheckHandle};
+use ra_flycheck::{CheckTask, FlycheckConfig, FlycheckHandle};
 use ra_ide::{Analysis, AnalysisChange, AnalysisHost, CrateGraph, FileId};
 use ra_project_model::{CargoWorkspace, ProcMacroClient, ProjectWorkspace, Target};
 use stdx::format_to;
@@ -30,12 +30,15 @@ use rustc_hash::{FxHashMap, FxHashSet};
 fn create_flycheck(
     workspaces: &[ProjectWorkspace],
     config: &FlycheckConfig,
-) -> Option<FlycheckHandle> {
+) -> Option<(FlycheckHandle, Receiver<CheckTask>)> {
     // FIXME: Figure out the multi-workspace situation
     workspaces.iter().find_map(move |w| match w {
         ProjectWorkspace::Cargo { cargo, .. } => {
+            let (sender, receiver) = unbounded();
+            let sender = Box::new(move |msg| sender.send(msg).unwrap());
             let cargo_project_root = cargo.workspace_root().to_path_buf();
-            Some(FlycheckHandle::spawn(config.clone(), cargo_project_root.into()))
+            let flycheck = FlycheckHandle::spawn(config.clone(), cargo_project_root.into(), sender);
+            Some((flycheck, receiver))
         }
         ProjectWorkspace::Json { .. } => {
             log::warn!("Cargo check watching only supported for cargo workspaces, disabling");
@@ -66,7 +69,7 @@ pub(crate) struct GlobalState {
     pub(crate) analysis_host: AnalysisHost,
     pub(crate) loader: Box<dyn vfs::loader::Handle>,
     pub(crate) task_receiver: Receiver<vfs::loader::Message>,
-    pub(crate) flycheck: Option<FlycheckHandle>,
+    pub(crate) flycheck: Option<(FlycheckHandle, Receiver<CheckTask>)>,
     pub(crate) diagnostics: DiagnosticCollection,
     pub(crate) mem_docs: FxHashSet<VfsPath>,
     pub(crate) vfs: Arc<RwLock<(vfs::Vfs, FxHashMap<FileId, LineEndings>)>>,

crates/rust-analyzer/src/main_loop.rs

@@ -136,7 +136,7 @@ pub fn main_loop(config: Config, connection: Connection) -> Result<()> {
                     Ok(task) => Event::Vfs(task),
                     Err(RecvError) => return Err("vfs died".into()),
                 },
-                recv(global_state.flycheck.as_ref().map_or(&never(), |it| &it.task_recv)) -> task => match task {
+                recv(global_state.flycheck.as_ref().map_or(&never(), |it| &it.1)) -> task => match task {
                     Ok(task) => Event::CheckWatcher(task),
                     Err(RecvError) => return Err("check watcher died".into()),
                 },
@@ -290,7 +290,7 @@ fn loop_turn(
 
     if became_ready {
         if let Some(flycheck) = &global_state.flycheck {
-            flycheck.update();
+            flycheck.0.update();
         }
     }
 
@@ -486,7 +486,7 @@ fn on_notification(
     let not = match notification_cast::<lsp_types::notification::DidSaveTextDocument>(not) {
         Ok(_params) => {
             if let Some(flycheck) = &global_state.flycheck {
-                flycheck.update();
+                flycheck.0.update();
             }
             return Ok(());
         }