From 4cbc902fcc9de79893779582dac01351d1137c7f Mon Sep 17 00:00:00 2001 From: Aleksey Kladov Date: Sat, 8 Dec 2018 19:30:35 +0300 Subject: grand module rename --- crates/ra_lsp_server/src/main_loop.rs | 495 ++++++++++++++++++++++++++++++ crates/ra_lsp_server/src/main_loop/mod.rs | 495 ------------------------------ 2 files changed, 495 insertions(+), 495 deletions(-) create mode 100644 crates/ra_lsp_server/src/main_loop.rs delete mode 100644 crates/ra_lsp_server/src/main_loop/mod.rs (limited to 'crates/ra_lsp_server/src') diff --git a/crates/ra_lsp_server/src/main_loop.rs b/crates/ra_lsp_server/src/main_loop.rs new file mode 100644 index 000000000..0e1878906 --- /dev/null +++ b/crates/ra_lsp_server/src/main_loop.rs @@ -0,0 +1,495 @@ +mod handlers; +mod subscriptions; + +use std::path::PathBuf; + +use crossbeam_channel::{unbounded, select, Receiver, Sender}; +use gen_lsp_server::{ + handle_shutdown, ErrorCode, RawMessage, RawNotification, RawRequest, RawResponse, +}; +use languageserver_types::NumberOrString; +use ra_analysis::{Canceled, FileId, LibraryData}; +use rayon::{self, ThreadPool}; +use rustc_hash::FxHashSet; +use serde::{de::DeserializeOwned, Serialize}; +use failure::{format_err, bail}; +use failure_derive::Fail; + +use crate::{ + main_loop::subscriptions::Subscriptions, + project_model::{workspace_loader, CargoWorkspace}, + req, + server_world::{ServerWorld, ServerWorldState}, + thread_watcher::Worker, + vfs::{self, FileEvent}, + Result, +}; + +#[derive(Debug, Fail)] +#[fail( + display = "Language Server request failed with {}. ({})", + code, message +)] +pub struct LspError { + pub code: i32, + pub message: String, +} + +impl LspError { + pub fn new(code: i32, message: String) -> LspError { + LspError { code, message } + } +} + +#[derive(Debug)] +enum Task { + Respond(RawResponse), + Notify(RawNotification), +} + +pub fn main_loop( + internal_mode: bool, + root: PathBuf, + publish_decorations: bool, + msg_receiver: &Receiver, + msg_sender: &Sender, +) -> Result<()> { + let pool = rayon::ThreadPoolBuilder::new() + .num_threads(4) + .panic_handler(|_| log::error!("thread panicked :(")) + .build() + .unwrap(); + let (task_sender, task_receiver) = unbounded::(); + let (fs_worker, fs_watcher) = vfs::roots_loader(); + let (ws_worker, ws_watcher) = workspace_loader(); + + log::info!("server initialized, serving requests"); + let mut state = ServerWorldState::default(); + + let mut pending_requests = FxHashSet::default(); + let mut subs = Subscriptions::new(); + let main_res = main_loop_inner( + internal_mode, + publish_decorations, + root, + &pool, + msg_sender, + msg_receiver, + task_sender, + task_receiver.clone(), + fs_worker, + ws_worker, + &mut state, + &mut pending_requests, + &mut subs, + ); + + log::info!("waiting for tasks to finish..."); + task_receiver.for_each(|task| on_task(task, msg_sender, &mut pending_requests)); + log::info!("...tasks have finished"); + log::info!("joining threadpool..."); + drop(pool); + log::info!("...threadpool has finished"); + + let fs_res = fs_watcher.stop(); + let ws_res = ws_watcher.stop(); + + main_res?; + fs_res?; + ws_res?; + + Ok(()) +} + +fn main_loop_inner( + internal_mode: bool, + publish_decorations: bool, + ws_root: PathBuf, + pool: &ThreadPool, + msg_sender: &Sender, + msg_receiver: &Receiver, + task_sender: Sender, + task_receiver: Receiver, + fs_worker: Worker)>, + ws_worker: Worker>, + state: &mut ServerWorldState, + pending_requests: &mut FxHashSet, + subs: &mut Subscriptions, +) -> Result<()> { + let (libdata_sender, libdata_receiver) = unbounded(); + ws_worker.send(ws_root.clone()); + fs_worker.send(ws_root.clone()); + loop { + #[derive(Debug)] + enum Event { + Msg(RawMessage), + Task(Task), + Fs(PathBuf, Vec), + Ws(Result), + Lib(LibraryData), + } + log::trace!("selecting"); + let event = select! { + recv(msg_receiver, msg) => match msg { + Some(msg) => Event::Msg(msg), + None => bail!("client exited without shutdown"), + }, + recv(task_receiver, task) => Event::Task(task.unwrap()), + recv(fs_worker.out, events) => match events { + None => bail!("roots watcher died"), + Some((pb, events)) => Event::Fs(pb, events), + } + recv(ws_worker.out, ws) => match ws { + None => bail!("workspace watcher died"), + Some(ws) => Event::Ws(ws), + } + recv(libdata_receiver, data) => Event::Lib(data.unwrap()) + }; + let mut state_changed = false; + match event { + Event::Task(task) => on_task(task, msg_sender, pending_requests), + Event::Fs(root, events) => { + log::info!("fs change, {}, {} events", root.display(), events.len()); + if root == ws_root { + state.apply_fs_changes(events); + } else { + let (files, resolver) = state.events_to_files(events); + let sender = libdata_sender.clone(); + pool.spawn(move || { + let start = ::std::time::Instant::now(); + log::info!("indexing {} ... ", root.display()); + let data = LibraryData::prepare(files, resolver); + log::info!("indexed {:?} {}", start.elapsed(), root.display()); + sender.send(data); + }); + } + state_changed = true; + } + Event::Ws(ws) => match ws { + Ok(ws) => { + let workspaces = vec![ws]; + feedback(internal_mode, "workspace loaded", msg_sender); + for ws in workspaces.iter() { + // Add each library as constant input. If library is + // within the workspace, don't treat it as a library. + // + // HACK: If source roots are nested, pick the outer one. + + let mut roots = ws + .packages() + .filter(|pkg| !pkg.is_member(ws)) + .filter_map(|pkg| { + let root = pkg.root(ws).to_path_buf(); + if root.starts_with(&ws_root) { + None + } else { + Some(root) + } + }) + .collect::>(); + roots.sort_by_key(|it| it.as_os_str().len()); + let unique = roots + .iter() + .enumerate() + .filter(|&(idx, long)| { + !roots[..idx].iter().any(|short| long.starts_with(short)) + }) + .map(|(_idx, root)| root); + + for root in unique { + log::debug!("sending root, {}", root.display()); + fs_worker.send(root.to_owned()); + } + } + state.set_workspaces(workspaces); + state_changed = true; + } + Err(e) => log::warn!("loading workspace failed: {}", e), + }, + Event::Lib(lib) => { + feedback(internal_mode, "library loaded", msg_sender); + state.add_lib(lib); + } + Event::Msg(msg) => match msg { + RawMessage::Request(req) => { + let req = match handle_shutdown(req, msg_sender) { + Some(req) => req, + None => return Ok(()), + }; + match on_request(state, pending_requests, pool, &task_sender, req)? { + None => (), + Some(req) => { + log::error!("unknown request: {:?}", req); + let resp = RawResponse::err( + req.id, + ErrorCode::MethodNotFound as i32, + "unknown request".to_string(), + ); + msg_sender.send(RawMessage::Response(resp)) + } + } + } + RawMessage::Notification(not) => { + on_notification(msg_sender, state, pending_requests, subs, not)?; + state_changed = true; + } + RawMessage::Response(resp) => log::error!("unexpected response: {:?}", resp), + }, + }; + + if state_changed { + update_file_notifications_on_threadpool( + pool, + state.snapshot(), + publish_decorations, + task_sender.clone(), + subs.subscriptions(), + ) + } + } +} + +fn on_task(task: Task, msg_sender: &Sender, pending_requests: &mut FxHashSet) { + match task { + Task::Respond(response) => { + if pending_requests.remove(&response.id) { + msg_sender.send(RawMessage::Response(response)) + } + } + Task::Notify(n) => msg_sender.send(RawMessage::Notification(n)), + } +} + +fn on_request( + world: &mut ServerWorldState, + pending_requests: &mut FxHashSet, + pool: &ThreadPool, + sender: &Sender, + req: RawRequest, +) -> Result> { + let mut pool_dispatcher = PoolDispatcher { + req: Some(req), + res: None, + pool, + world, + sender, + }; + let req = pool_dispatcher + .on::(handlers::handle_syntax_tree)? + .on::(handlers::handle_extend_selection)? + .on::(handlers::handle_find_matching_brace)? + .on::(handlers::handle_join_lines)? + .on::(handlers::handle_on_enter)? + .on::(handlers::handle_on_type_formatting)? + .on::(handlers::handle_document_symbol)? + .on::(handlers::handle_workspace_symbol)? + .on::(handlers::handle_goto_definition)? + .on::(handlers::handle_parent_module)? + .on::(handlers::handle_runnables)? + .on::(handlers::handle_decorations)? + .on::(handlers::handle_completion)? + .on::(handlers::handle_code_action)? + .on::(handlers::handle_folding_range)? + .on::(handlers::handle_signature_help)? + .on::(handlers::handle_hover)? + .on::(handlers::handle_prepare_rename)? + .on::(handlers::handle_rename)? + .on::(handlers::handle_references)? + .finish(); + match req { + Ok(id) => { + let inserted = pending_requests.insert(id); + assert!(inserted, "duplicate request: {}", id); + Ok(None) + } + Err(req) => Ok(Some(req)), + } +} + +fn on_notification( + msg_sender: &Sender, + state: &mut ServerWorldState, + pending_requests: &mut FxHashSet, + subs: &mut Subscriptions, + not: RawNotification, +) -> Result<()> { + let not = match not.cast::() { + Ok(params) => { + let id = match params.id { + NumberOrString::Number(id) => id, + NumberOrString::String(id) => { + panic!("string id's not supported: {:?}", id); + } + }; + pending_requests.remove(&id); + return Ok(()); + } + Err(not) => not, + }; + let not = match not.cast::() { + Ok(params) => { + let uri = params.text_document.uri; + let path = uri + .to_file_path() + .map_err(|()| format_err!("invalid uri: {}", uri))?; + let file_id = state.add_mem_file(path, params.text_document.text); + subs.add_sub(file_id); + return Ok(()); + } + Err(not) => not, + }; + let not = match not.cast::() { + Ok(mut params) => { + let uri = params.text_document.uri; + let path = uri + .to_file_path() + .map_err(|()| format_err!("invalid uri: {}", uri))?; + let text = params + .content_changes + .pop() + .ok_or_else(|| format_err!("empty changes"))? + .text; + state.change_mem_file(path.as_path(), text)?; + return Ok(()); + } + Err(not) => not, + }; + let not = match not.cast::() { + Ok(params) => { + let uri = params.text_document.uri; + let path = uri + .to_file_path() + .map_err(|()| format_err!("invalid uri: {}", uri))?; + let file_id = state.remove_mem_file(path.as_path())?; + subs.remove_sub(file_id); + let params = req::PublishDiagnosticsParams { + uri, + diagnostics: Vec::new(), + }; + let not = RawNotification::new::(¶ms); + msg_sender.send(RawMessage::Notification(not)); + return Ok(()); + } + Err(not) => not, + }; + log::error!("unhandled notification: {:?}", not); + Ok(()) +} + +struct PoolDispatcher<'a> { + req: Option, + res: Option, + pool: &'a ThreadPool, + world: &'a ServerWorldState, + sender: &'a Sender, +} + +impl<'a> PoolDispatcher<'a> { + fn on<'b, R>( + &'b mut self, + f: fn(ServerWorld, R::Params) -> Result, + ) -> Result<&'b mut Self> + where + R: req::Request, + R::Params: DeserializeOwned + Send + 'static, + R::Result: Serialize + 'static, + { + let req = match self.req.take() { + None => return Ok(self), + Some(req) => req, + }; + match req.cast::() { + Ok((id, params)) => { + let world = self.world.snapshot(); + let sender = self.sender.clone(); + self.pool.spawn(move || { + let resp = match f(world, params) { + Ok(resp) => RawResponse::ok::(id, &resp), + Err(e) => match e.downcast::() { + Ok(lsp_error) => { + RawResponse::err(id, lsp_error.code, lsp_error.message) + } + Err(e) => { + if is_canceled(&e) { + RawResponse::err( + id, + ErrorCode::RequestCancelled as i32, + e.to_string(), + ) + } else { + RawResponse::err( + id, + ErrorCode::InternalError as i32, + format!("{}\n{}", e, e.backtrace()), + ) + } + } + }, + }; + let task = Task::Respond(resp); + sender.send(task); + }); + self.res = Some(id); + } + Err(req) => self.req = Some(req), + } + Ok(self) + } + + fn finish(&mut self) -> ::std::result::Result { + match (self.res.take(), self.req.take()) { + (Some(res), None) => Ok(res), + (None, Some(req)) => Err(req), + _ => unreachable!(), + } + } +} + +fn update_file_notifications_on_threadpool( + pool: &ThreadPool, + world: ServerWorld, + publish_decorations: bool, + sender: Sender, + subscriptions: Vec, +) { + pool.spawn(move || { + for file_id in subscriptions { + match handlers::publish_diagnostics(&world, file_id) { + Err(e) => { + if !is_canceled(&e) { + log::error!("failed to compute diagnostics: {:?}", e); + } + } + Ok(params) => { + let not = RawNotification::new::(¶ms); + sender.send(Task::Notify(not)); + } + } + if publish_decorations { + match handlers::publish_decorations(&world, file_id) { + Err(e) => { + if !is_canceled(&e) { + log::error!("failed to compute decorations: {:?}", e); + } + } + Ok(params) => { + let not = RawNotification::new::(¶ms); + sender.send(Task::Notify(not)) + } + } + } + } + }); +} + +fn feedback(intrnal_mode: bool, msg: &str, sender: &Sender) { + if !intrnal_mode { + return; + } + let not = RawNotification::new::(&msg.to_string()); + sender.send(RawMessage::Notification(not)); +} + +fn is_canceled(e: &failure::Error) -> bool { + e.downcast_ref::().is_some() +} diff --git a/crates/ra_lsp_server/src/main_loop/mod.rs b/crates/ra_lsp_server/src/main_loop/mod.rs deleted file mode 100644 index 0e1878906..000000000 --- a/crates/ra_lsp_server/src/main_loop/mod.rs +++ /dev/null @@ -1,495 +0,0 @@ -mod handlers; -mod subscriptions; - -use std::path::PathBuf; - -use crossbeam_channel::{unbounded, select, Receiver, Sender}; -use gen_lsp_server::{ - handle_shutdown, ErrorCode, RawMessage, RawNotification, RawRequest, RawResponse, -}; -use languageserver_types::NumberOrString; -use ra_analysis::{Canceled, FileId, LibraryData}; -use rayon::{self, ThreadPool}; -use rustc_hash::FxHashSet; -use serde::{de::DeserializeOwned, Serialize}; -use failure::{format_err, bail}; -use failure_derive::Fail; - -use crate::{ - main_loop::subscriptions::Subscriptions, - project_model::{workspace_loader, CargoWorkspace}, - req, - server_world::{ServerWorld, ServerWorldState}, - thread_watcher::Worker, - vfs::{self, FileEvent}, - Result, -}; - -#[derive(Debug, Fail)] -#[fail( - display = "Language Server request failed with {}. ({})", - code, message -)] -pub struct LspError { - pub code: i32, - pub message: String, -} - -impl LspError { - pub fn new(code: i32, message: String) -> LspError { - LspError { code, message } - } -} - -#[derive(Debug)] -enum Task { - Respond(RawResponse), - Notify(RawNotification), -} - -pub fn main_loop( - internal_mode: bool, - root: PathBuf, - publish_decorations: bool, - msg_receiver: &Receiver, - msg_sender: &Sender, -) -> Result<()> { - let pool = rayon::ThreadPoolBuilder::new() - .num_threads(4) - .panic_handler(|_| log::error!("thread panicked :(")) - .build() - .unwrap(); - let (task_sender, task_receiver) = unbounded::(); - let (fs_worker, fs_watcher) = vfs::roots_loader(); - let (ws_worker, ws_watcher) = workspace_loader(); - - log::info!("server initialized, serving requests"); - let mut state = ServerWorldState::default(); - - let mut pending_requests = FxHashSet::default(); - let mut subs = Subscriptions::new(); - let main_res = main_loop_inner( - internal_mode, - publish_decorations, - root, - &pool, - msg_sender, - msg_receiver, - task_sender, - task_receiver.clone(), - fs_worker, - ws_worker, - &mut state, - &mut pending_requests, - &mut subs, - ); - - log::info!("waiting for tasks to finish..."); - task_receiver.for_each(|task| on_task(task, msg_sender, &mut pending_requests)); - log::info!("...tasks have finished"); - log::info!("joining threadpool..."); - drop(pool); - log::info!("...threadpool has finished"); - - let fs_res = fs_watcher.stop(); - let ws_res = ws_watcher.stop(); - - main_res?; - fs_res?; - ws_res?; - - Ok(()) -} - -fn main_loop_inner( - internal_mode: bool, - publish_decorations: bool, - ws_root: PathBuf, - pool: &ThreadPool, - msg_sender: &Sender, - msg_receiver: &Receiver, - task_sender: Sender, - task_receiver: Receiver, - fs_worker: Worker)>, - ws_worker: Worker>, - state: &mut ServerWorldState, - pending_requests: &mut FxHashSet, - subs: &mut Subscriptions, -) -> Result<()> { - let (libdata_sender, libdata_receiver) = unbounded(); - ws_worker.send(ws_root.clone()); - fs_worker.send(ws_root.clone()); - loop { - #[derive(Debug)] - enum Event { - Msg(RawMessage), - Task(Task), - Fs(PathBuf, Vec), - Ws(Result), - Lib(LibraryData), - } - log::trace!("selecting"); - let event = select! { - recv(msg_receiver, msg) => match msg { - Some(msg) => Event::Msg(msg), - None => bail!("client exited without shutdown"), - }, - recv(task_receiver, task) => Event::Task(task.unwrap()), - recv(fs_worker.out, events) => match events { - None => bail!("roots watcher died"), - Some((pb, events)) => Event::Fs(pb, events), - } - recv(ws_worker.out, ws) => match ws { - None => bail!("workspace watcher died"), - Some(ws) => Event::Ws(ws), - } - recv(libdata_receiver, data) => Event::Lib(data.unwrap()) - }; - let mut state_changed = false; - match event { - Event::Task(task) => on_task(task, msg_sender, pending_requests), - Event::Fs(root, events) => { - log::info!("fs change, {}, {} events", root.display(), events.len()); - if root == ws_root { - state.apply_fs_changes(events); - } else { - let (files, resolver) = state.events_to_files(events); - let sender = libdata_sender.clone(); - pool.spawn(move || { - let start = ::std::time::Instant::now(); - log::info!("indexing {} ... ", root.display()); - let data = LibraryData::prepare(files, resolver); - log::info!("indexed {:?} {}", start.elapsed(), root.display()); - sender.send(data); - }); - } - state_changed = true; - } - Event::Ws(ws) => match ws { - Ok(ws) => { - let workspaces = vec![ws]; - feedback(internal_mode, "workspace loaded", msg_sender); - for ws in workspaces.iter() { - // Add each library as constant input. If library is - // within the workspace, don't treat it as a library. - // - // HACK: If source roots are nested, pick the outer one. - - let mut roots = ws - .packages() - .filter(|pkg| !pkg.is_member(ws)) - .filter_map(|pkg| { - let root = pkg.root(ws).to_path_buf(); - if root.starts_with(&ws_root) { - None - } else { - Some(root) - } - }) - .collect::>(); - roots.sort_by_key(|it| it.as_os_str().len()); - let unique = roots - .iter() - .enumerate() - .filter(|&(idx, long)| { - !roots[..idx].iter().any(|short| long.starts_with(short)) - }) - .map(|(_idx, root)| root); - - for root in unique { - log::debug!("sending root, {}", root.display()); - fs_worker.send(root.to_owned()); - } - } - state.set_workspaces(workspaces); - state_changed = true; - } - Err(e) => log::warn!("loading workspace failed: {}", e), - }, - Event::Lib(lib) => { - feedback(internal_mode, "library loaded", msg_sender); - state.add_lib(lib); - } - Event::Msg(msg) => match msg { - RawMessage::Request(req) => { - let req = match handle_shutdown(req, msg_sender) { - Some(req) => req, - None => return Ok(()), - }; - match on_request(state, pending_requests, pool, &task_sender, req)? { - None => (), - Some(req) => { - log::error!("unknown request: {:?}", req); - let resp = RawResponse::err( - req.id, - ErrorCode::MethodNotFound as i32, - "unknown request".to_string(), - ); - msg_sender.send(RawMessage::Response(resp)) - } - } - } - RawMessage::Notification(not) => { - on_notification(msg_sender, state, pending_requests, subs, not)?; - state_changed = true; - } - RawMessage::Response(resp) => log::error!("unexpected response: {:?}", resp), - }, - }; - - if state_changed { - update_file_notifications_on_threadpool( - pool, - state.snapshot(), - publish_decorations, - task_sender.clone(), - subs.subscriptions(), - ) - } - } -} - -fn on_task(task: Task, msg_sender: &Sender, pending_requests: &mut FxHashSet) { - match task { - Task::Respond(response) => { - if pending_requests.remove(&response.id) { - msg_sender.send(RawMessage::Response(response)) - } - } - Task::Notify(n) => msg_sender.send(RawMessage::Notification(n)), - } -} - -fn on_request( - world: &mut ServerWorldState, - pending_requests: &mut FxHashSet, - pool: &ThreadPool, - sender: &Sender, - req: RawRequest, -) -> Result> { - let mut pool_dispatcher = PoolDispatcher { - req: Some(req), - res: None, - pool, - world, - sender, - }; - let req = pool_dispatcher - .on::(handlers::handle_syntax_tree)? - .on::(handlers::handle_extend_selection)? - .on::(handlers::handle_find_matching_brace)? - .on::(handlers::handle_join_lines)? - .on::(handlers::handle_on_enter)? - .on::(handlers::handle_on_type_formatting)? - .on::(handlers::handle_document_symbol)? - .on::(handlers::handle_workspace_symbol)? - .on::(handlers::handle_goto_definition)? - .on::(handlers::handle_parent_module)? - .on::(handlers::handle_runnables)? - .on::(handlers::handle_decorations)? - .on::(handlers::handle_completion)? - .on::(handlers::handle_code_action)? - .on::(handlers::handle_folding_range)? - .on::(handlers::handle_signature_help)? - .on::(handlers::handle_hover)? - .on::(handlers::handle_prepare_rename)? - .on::(handlers::handle_rename)? - .on::(handlers::handle_references)? - .finish(); - match req { - Ok(id) => { - let inserted = pending_requests.insert(id); - assert!(inserted, "duplicate request: {}", id); - Ok(None) - } - Err(req) => Ok(Some(req)), - } -} - -fn on_notification( - msg_sender: &Sender, - state: &mut ServerWorldState, - pending_requests: &mut FxHashSet, - subs: &mut Subscriptions, - not: RawNotification, -) -> Result<()> { - let not = match not.cast::() { - Ok(params) => { - let id = match params.id { - NumberOrString::Number(id) => id, - NumberOrString::String(id) => { - panic!("string id's not supported: {:?}", id); - } - }; - pending_requests.remove(&id); - return Ok(()); - } - Err(not) => not, - }; - let not = match not.cast::() { - Ok(params) => { - let uri = params.text_document.uri; - let path = uri - .to_file_path() - .map_err(|()| format_err!("invalid uri: {}", uri))?; - let file_id = state.add_mem_file(path, params.text_document.text); - subs.add_sub(file_id); - return Ok(()); - } - Err(not) => not, - }; - let not = match not.cast::() { - Ok(mut params) => { - let uri = params.text_document.uri; - let path = uri - .to_file_path() - .map_err(|()| format_err!("invalid uri: {}", uri))?; - let text = params - .content_changes - .pop() - .ok_or_else(|| format_err!("empty changes"))? - .text; - state.change_mem_file(path.as_path(), text)?; - return Ok(()); - } - Err(not) => not, - }; - let not = match not.cast::() { - Ok(params) => { - let uri = params.text_document.uri; - let path = uri - .to_file_path() - .map_err(|()| format_err!("invalid uri: {}", uri))?; - let file_id = state.remove_mem_file(path.as_path())?; - subs.remove_sub(file_id); - let params = req::PublishDiagnosticsParams { - uri, - diagnostics: Vec::new(), - }; - let not = RawNotification::new::(¶ms); - msg_sender.send(RawMessage::Notification(not)); - return Ok(()); - } - Err(not) => not, - }; - log::error!("unhandled notification: {:?}", not); - Ok(()) -} - -struct PoolDispatcher<'a> { - req: Option, - res: Option, - pool: &'a ThreadPool, - world: &'a ServerWorldState, - sender: &'a Sender, -} - -impl<'a> PoolDispatcher<'a> { - fn on<'b, R>( - &'b mut self, - f: fn(ServerWorld, R::Params) -> Result, - ) -> Result<&'b mut Self> - where - R: req::Request, - R::Params: DeserializeOwned + Send + 'static, - R::Result: Serialize + 'static, - { - let req = match self.req.take() { - None => return Ok(self), - Some(req) => req, - }; - match req.cast::() { - Ok((id, params)) => { - let world = self.world.snapshot(); - let sender = self.sender.clone(); - self.pool.spawn(move || { - let resp = match f(world, params) { - Ok(resp) => RawResponse::ok::(id, &resp), - Err(e) => match e.downcast::() { - Ok(lsp_error) => { - RawResponse::err(id, lsp_error.code, lsp_error.message) - } - Err(e) => { - if is_canceled(&e) { - RawResponse::err( - id, - ErrorCode::RequestCancelled as i32, - e.to_string(), - ) - } else { - RawResponse::err( - id, - ErrorCode::InternalError as i32, - format!("{}\n{}", e, e.backtrace()), - ) - } - } - }, - }; - let task = Task::Respond(resp); - sender.send(task); - }); - self.res = Some(id); - } - Err(req) => self.req = Some(req), - } - Ok(self) - } - - fn finish(&mut self) -> ::std::result::Result { - match (self.res.take(), self.req.take()) { - (Some(res), None) => Ok(res), - (None, Some(req)) => Err(req), - _ => unreachable!(), - } - } -} - -fn update_file_notifications_on_threadpool( - pool: &ThreadPool, - world: ServerWorld, - publish_decorations: bool, - sender: Sender, - subscriptions: Vec, -) { - pool.spawn(move || { - for file_id in subscriptions { - match handlers::publish_diagnostics(&world, file_id) { - Err(e) => { - if !is_canceled(&e) { - log::error!("failed to compute diagnostics: {:?}", e); - } - } - Ok(params) => { - let not = RawNotification::new::(¶ms); - sender.send(Task::Notify(not)); - } - } - if publish_decorations { - match handlers::publish_decorations(&world, file_id) { - Err(e) => { - if !is_canceled(&e) { - log::error!("failed to compute decorations: {:?}", e); - } - } - Ok(params) => { - let not = RawNotification::new::(¶ms); - sender.send(Task::Notify(not)) - } - } - } - } - }); -} - -fn feedback(intrnal_mode: bool, msg: &str, sender: &Sender) { - if !intrnal_mode { - return; - } - let not = RawNotification::new::(&msg.to_string()); - sender.send(RawMessage::Notification(not)); -} - -fn is_canceled(e: &failure::Error) -> bool { - e.downcast_ref::().is_some() -} -- cgit v1.2.3