crates/ra_lsp_server/src/main_loop.rs

   1 //! The main loop of `ra_lsp_server` responsible for dispatching LSP requests/replies and
   2 //! notifications back to the client.
   3
   4 mod handlers;
   5 mod subscriptions;
   6 pub(crate) mod pending_requests;
   7
   8 use std::{error::Error, fmt, panic, path::PathBuf, sync::Arc, time::Instant};
   9
  10 use crossbeam_channel::{select, unbounded, RecvError, Sender};
  11 use lsp_server::{Connection, ErrorCode, Message, Notification, Request, RequestId, Response};
  12 use lsp_types::{ClientCapabilities, NumberOrString};
  13 use ra_ide::{Canceled, FeatureFlags, FileId, LibraryData, SourceRootId};
  14 use ra_prof::profile;
  15 use ra_vfs::{VfsTask, Watch};
  16 use relative_path::RelativePathBuf;
  17 use rustc_hash::FxHashSet;
  18 use serde::{de::DeserializeOwned, Serialize};
  19 use threadpool::ThreadPool;
  20
  21 use crate::{
  22     cargo_check::CheckTask,
  23     main_loop::{
  24         pending_requests::{PendingRequest, PendingRequests},
  25         subscriptions::Subscriptions,
  26     },
  27     req,
  28     world::{Options, WorldSnapshot, WorldState},
  29     Result, ServerConfig,
  30 };
  31
  32 const THREADPOOL_SIZE: usize = 8;
  33 const MAX_IN_FLIGHT_LIBS: usize = THREADPOOL_SIZE - 3;
  34
  35 #[derive(Debug)]
  36 pub struct LspError {
  37     pub code: i32,
  38     pub message: String,
  39 }
  40
  41 impl LspError {
  42     pub fn new(code: i32, message: String) -> LspError {
  43         LspError { code, message }
  44     }
  45 }
  46
  47 impl fmt::Display for LspError {
  48     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
  49         write!(f, "Language Server request failed with {}. ({})", self.code, self.message)
  50     }
  51 }
  52
  53 impl Error for LspError {}
  54
  55 pub fn main_loop(
  56     ws_roots: Vec<PathBuf>,
  57     client_caps: ClientCapabilities,
  58     config: ServerConfig,
  59     connection: Connection,
  60 ) -> Result<()> {
  61     log::info!("server_config: {:#?}", config);
  62
  63     let mut loop_state = LoopState::default();
  64     let mut world_state = {
  65         // FIXME: support dynamic workspace loading.
  66         let workspaces = {
  67             let mut loaded_workspaces = Vec::new();
  68             for ws_root in &ws_roots {
  69                 let workspace = ra_project_model::ProjectWorkspace::discover_with_sysroot(
  70                     ws_root.as_path(),
  71                     config.with_sysroot,
  72                     &config.cargo_features,
  73                 );
  74                 match workspace {
  75                     Ok(workspace) => loaded_workspaces.push(workspace),
  76                     Err(e) => {
  77                         log::error!("loading workspace failed: {}", e);
  78
  79                         show_message(
  80                             req::MessageType::Error,
  81                             format!("rust-analyzer failed to load workspace: {}", e),
  82                             &connection.sender,
  83                         );
  84                     }
  85                 }
  86             }
  87             loaded_workspaces
  88         };
  89
  90         let globs = config
  91             .exclude_globs
  92             .iter()
  93             .map(|glob| ra_vfs_glob::Glob::new(glob))
  94             .collect::<std::result::Result<Vec<_>, _>>()?;
  95
  96         if config.use_client_watching {
  97             let registration_options = req::DidChangeWatchedFilesRegistrationOptions {
  98                 watchers: workspaces
  99                     .iter()
 100                     .flat_map(|ws| ws.to_roots())
 101                     .filter(|root| root.is_member())
 102                     .map(|root| format!("{}/**/*.rs", root.path().display()))
 103                     .map(|glob_pattern| req::FileSystemWatcher { glob_pattern, kind: None })
 104                     .collect(),
 105             };
 106             let registration = req::Registration {
 107                 id: "file-watcher".to_string(),
 108                 method: "workspace/didChangeWatchedFiles".to_string(),
 109                 register_options: Some(serde_json::to_value(registration_options).unwrap()),
 110             };
 111             let params = req::RegistrationParams { registrations: vec![registration] };
 112             let request =
 113                 request_new::<req::RegisterCapability>(loop_state.next_request_id(), params);
 114             connection.sender.send(request.into()).unwrap();
 115         }
 116
 117         let options = {
 118             let text_document_caps = client_caps.text_document.as_ref();
 119             Options {
 120                 publish_decorations: config.publish_decorations,
 121                 supports_location_link: text_document_caps
 122                     .and_then(|it| it.definition)
 123                     .and_then(|it| it.link_support)
 124                     .unwrap_or(false),
 125                 line_folding_only: text_document_caps
 126                     .and_then(|it| it.folding_range.as_ref())
 127                     .and_then(|it| it.line_folding_only)
 128                     .unwrap_or(false),
 129                 max_inlay_hint_length: config.max_inlay_hint_length,
 130                 cargo_check_enable: config.cargo_check_enable,
 131                 cargo_check_command: config.cargo_check_command,
 132                 cargo_check_args: config.cargo_check_args,
 133             }
 134         };
 135
 136         let feature_flags = {
 137             let mut ff = FeatureFlags::default();
 138             for (flag, value) in config.feature_flags {
 139                 if ff.set(flag.as_str(), value).is_err() {
 140                     log::error!("unknown feature flag: {:?}", flag);
 141                     show_message(
 142                         req::MessageType::Error,
 143                         format!("unknown feature flag: {:?}", flag),
 144                         &connection.sender,
 145                     );
 146                 }
 147             }
 148             ff
 149         };
 150         log::info!("feature_flags: {:#?}", feature_flags);
 151
 152         WorldState::new(
 153             ws_roots,
 154             workspaces,
 155             config.lru_capacity,
 156             &globs,
 157             Watch(!config.use_client_watching),
 158             options,
 159             feature_flags,
 160         )
 161     };
 162
 163     let pool = ThreadPool::new(THREADPOOL_SIZE);
 164     let (task_sender, task_receiver) = unbounded::<Task>();
 165     let (libdata_sender, libdata_receiver) = unbounded::<LibraryData>();
 166
 167     log::info!("server initialized, serving requests");
 168     {
 169         let task_sender = task_sender;
 170         let libdata_sender = libdata_sender;
 171         loop {
 172             log::trace!("selecting");
 173             let event = select! {
 174                 recv(&connection.receiver) -> msg => match msg {
 175                     Ok(msg) => Event::Msg(msg),
 176                     Err(RecvError) => Err("client exited without shutdown")?,
 177                 },
 178                 recv(task_receiver) -> task => Event::Task(task.unwrap()),
 179                 recv(world_state.task_receiver) -> task => match task {
 180                     Ok(task) => Event::Vfs(task),
 181                     Err(RecvError) => Err("vfs died")?,
 182                 },
 183                 recv(libdata_receiver) -> data => Event::Lib(data.unwrap()),
 184                 recv(world_state.check_watcher.task_recv) -> task => Event::CheckWatcher(task.unwrap())
 185             };
 186             if let Event::Msg(Message::Request(req)) = &event {
 187                 if connection.handle_shutdown(&req)? {
 188                     break;
 189                 };
 190             }
 191             loop_turn(
 192                 &pool,
 193                 &task_sender,
 194                 &libdata_sender,
 195                 &connection,
 196                 &mut world_state,
 197                 &mut loop_state,
 198                 event,
 199             )?;
 200         }
 201     }
 202
 203     log::info!("waiting for tasks to finish...");
 204     task_receiver.into_iter().for_each(|task| {
 205         on_task(task, &connection.sender, &mut loop_state.pending_requests, &mut world_state)
 206     });
 207     libdata_receiver.into_iter().for_each(drop);
 208     log::info!("...tasks have finished");
 209     log::info!("joining threadpool...");
 210     drop(pool);
 211     log::info!("...threadpool has finished");
 212
 213     let vfs = Arc::try_unwrap(world_state.vfs).expect("all snapshots should be dead");
 214     drop(vfs);
 215
 216     Ok(())
 217 }
 218
 219 #[derive(Debug)]
 220 enum Task {
 221     Respond(Response),
 222     Notify(Notification),
 223 }
 224
 225 enum Event {
 226     Msg(Message),
 227     Task(Task),
 228     Vfs(VfsTask),
 229     Lib(LibraryData),
 230     CheckWatcher(CheckTask),
 231 }
 232
 233 impl fmt::Debug for Event {
 234     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
 235         let debug_verbose_not = |not: &Notification, f: &mut fmt::Formatter| {
 236             f.debug_struct("Notification").field("method", &not.method).finish()
 237         };
 238
 239         match self {
 240             Event::Msg(Message::Notification(not)) => {
 241                 if notification_is::<req::DidOpenTextDocument>(not)
 242                     || notification_is::<req::DidChangeTextDocument>(not)
 243                 {
 244                     return debug_verbose_not(not, f);
 245                 }
 246             }
 247             Event::Task(Task::Notify(not)) => {
 248                 if notification_is::<req::PublishDecorations>(not)
 249                     || notification_is::<req::PublishDiagnostics>(not)
 250                 {
 251                     return debug_verbose_not(not, f);
 252                 }
 253             }
 254             Event::Task(Task::Respond(resp)) => {
 255                 return f
 256                     .debug_struct("Response")
 257                     .field("id", &resp.id)
 258                     .field("error", &resp.error)
 259                     .finish();
 260             }
 261             _ => (),
 262         }
 263         match self {
 264             Event::Msg(it) => fmt::Debug::fmt(it, f),
 265             Event::Task(it) => fmt::Debug::fmt(it, f),
 266             Event::Vfs(it) => fmt::Debug::fmt(it, f),
 267             Event::Lib(it) => fmt::Debug::fmt(it, f),
 268             Event::CheckWatcher(it) => fmt::Debug::fmt(it, f),
 269         }
 270     }
 271 }
 272
 273 #[derive(Debug, Default)]
 274 struct LoopState {
 275     next_request_id: u64,
 276     pending_responses: FxHashSet<RequestId>,
 277     pending_requests: PendingRequests,
 278     subscriptions: Subscriptions,
 279     // We try not to index more than MAX_IN_FLIGHT_LIBS libraries at the same
 280     // time to always have a thread ready to react to input.
 281     in_flight_libraries: usize,
 282     pending_libraries: Vec<(SourceRootId, Vec<(FileId, RelativePathBuf, Arc<String>)>)>,
 283     workspace_loaded: bool,
 284 }
 285
 286 impl LoopState {
 287     fn next_request_id(&mut self) -> RequestId {
 288         self.next_request_id += 1;
 289         let res: RequestId = self.next_request_id.into();
 290         let inserted = self.pending_responses.insert(res.clone());
 291         assert!(inserted);
 292         res
 293     }
 294 }
 295
 296 fn loop_turn(
 297     pool: &ThreadPool,
 298     task_sender: &Sender<Task>,
 299     libdata_sender: &Sender<LibraryData>,
 300     connection: &Connection,
 301     world_state: &mut WorldState,
 302     loop_state: &mut LoopState,
 303     event: Event,
 304 ) -> Result<()> {
 305     let loop_start = Instant::now();
 306
 307     // NOTE: don't count blocking select! call as a loop-turn time
 308     let _p = profile("main_loop_inner/loop-turn");
 309     log::info!("loop turn = {:?}", event);
 310     let queue_count = pool.queued_count();
 311     if queue_count > 0 {
 312         log::info!("queued count = {}", queue_count);
 313     }
 314
 315     match event {
 316         Event::Task(task) => {
 317             on_task(task, &connection.sender, &mut loop_state.pending_requests, world_state);
 318             world_state.maybe_collect_garbage();
 319         }
 320         Event::Vfs(task) => {
 321             world_state.vfs.write().handle_task(task);
 322         }
 323         Event::Lib(lib) => {
 324             world_state.add_lib(lib);
 325             world_state.maybe_collect_garbage();
 326             loop_state.in_flight_libraries -= 1;
 327         }
 328         Event::CheckWatcher(task) => match task {
 329             CheckTask::Update(uri) => {
 330                 // We manually send a diagnostic update when the watcher asks
 331                 // us to, to avoid the issue of having to change the file to
 332                 // receive updated diagnostics.
 333                 let path = uri.to_file_path().map_err(|()| format!("invalid uri: {}", uri))?;
 334                 if let Some(file_id) = world_state.vfs.read().path2file(&path) {
 335                     let params =
 336                         handlers::publish_diagnostics(&world_state.snapshot(), FileId(file_id.0))?;
 337                     let not = notification_new::<req::PublishDiagnostics>(params);
 338                     task_sender.send(Task::Notify(not)).unwrap();
 339                 }
 340             }
 341         },
 342         Event::Msg(msg) => match msg {
 343             Message::Request(req) => on_request(
 344                 world_state,
 345                 &mut loop_state.pending_requests,
 346                 pool,
 347                 task_sender,
 348                 &connection.sender,
 349                 loop_start,
 350                 req,
 351             )?,
 352             Message::Notification(not) => {
 353                 on_notification(
 354                     &connection.sender,
 355                     world_state,
 356                     &mut loop_state.pending_requests,
 357                     &mut loop_state.subscriptions,
 358                     not,
 359                 )?;
 360             }
 361             Message::Response(resp) => {
 362                 let removed = loop_state.pending_responses.remove(&resp.id);
 363                 if !removed {
 364                     log::error!("unexpected response: {:?}", resp)
 365                 }
 366             }
 367         },
 368     };
 369
 370     let mut state_changed = false;
 371     if let Some(changes) = world_state.process_changes() {
 372         state_changed = true;
 373         loop_state.pending_libraries.extend(changes);
 374     }
 375
 376     while loop_state.in_flight_libraries < MAX_IN_FLIGHT_LIBS
 377         && !loop_state.pending_libraries.is_empty()
 378     {
 379         let (root, files) = loop_state.pending_libraries.pop().unwrap();
 380         loop_state.in_flight_libraries += 1;
 381         let sender = libdata_sender.clone();
 382         pool.execute(move || {
 383             log::info!("indexing {:?} ... ", root);
 384             let _p = profile(&format!("indexed {:?}", root));
 385             let data = LibraryData::prepare(root, files);
 386             sender.send(data).unwrap();
 387         });
 388     }
 389
 390     if !loop_state.workspace_loaded
 391         && world_state.roots_to_scan == 0
 392         && loop_state.pending_libraries.is_empty()
 393         && loop_state.in_flight_libraries == 0
 394     {
 395         loop_state.workspace_loaded = true;
 396         let n_packages: usize = world_state.workspaces.iter().map(|it| it.n_packages()).sum();
 397         if world_state.feature_flags().get("notifications.workspace-loaded") {
 398             let msg = format!("workspace loaded, {} rust packages", n_packages);
 399             show_message(req::MessageType::Info, msg, &connection.sender);
 400         }
 401     }
 402
 403     if state_changed {
 404         update_file_notifications_on_threadpool(
 405             pool,
 406             world_state.snapshot(),
 407             world_state.options.publish_decorations,
 408             task_sender.clone(),
 409             loop_state.subscriptions.subscriptions(),
 410         )
 411     }
 412     Ok(())
 413 }
 414
 415 fn on_task(
 416     task: Task,
 417     msg_sender: &Sender<Message>,
 418     pending_requests: &mut PendingRequests,
 419     state: &mut WorldState,
 420 ) {
 421     match task {
 422         Task::Respond(response) => {
 423             if let Some(completed) = pending_requests.finish(&response.id) {
 424                 log::info!("handled req#{} in {:?}", completed.id, completed.duration);
 425                 state.complete_request(completed);
 426                 msg_sender.send(response.into()).unwrap();
 427             }
 428         }
 429         Task::Notify(n) => {
 430             msg_sender.send(n.into()).unwrap();
 431         }
 432     }
 433 }
 434
 435 fn on_request(
 436     world: &mut WorldState,
 437     pending_requests: &mut PendingRequests,
 438     pool: &ThreadPool,
 439     sender: &Sender<Task>,
 440     msg_sender: &Sender<Message>,
 441     request_received: Instant,
 442     req: Request,
 443 ) -> Result<()> {
 444     let mut pool_dispatcher = PoolDispatcher {
 445         req: Some(req),
 446         pool,
 447         world,
 448         sender,
 449         msg_sender,
 450         pending_requests,
 451         request_received,
 452     };
 453     pool_dispatcher
 454         .on_sync::<req::CollectGarbage>(|s, ()| Ok(s.collect_garbage()))?
 455         .on_sync::<req::JoinLines>(|s, p| handlers::handle_join_lines(s.snapshot(), p))?
 456         .on_sync::<req::OnEnter>(|s, p| handlers::handle_on_enter(s.snapshot(), p))?
 457         .on_sync::<req::SelectionRangeRequest>(|s, p| {
 458             handlers::handle_selection_range(s.snapshot(), p)
 459         })?
 460         .on_sync::<req::FindMatchingBrace>(|s, p| {
 461             handlers::handle_find_matching_brace(s.snapshot(), p)
 462         })?
 463         .on::<req::AnalyzerStatus>(handlers::handle_analyzer_status)?
 464         .on::<req::SyntaxTree>(handlers::handle_syntax_tree)?
 465         .on::<req::ExpandMacro>(handlers::handle_expand_macro)?
 466         .on::<req::OnTypeFormatting>(handlers::handle_on_type_formatting)?
 467         .on::<req::DocumentSymbolRequest>(handlers::handle_document_symbol)?
 468         .on::<req::WorkspaceSymbol>(handlers::handle_workspace_symbol)?
 469         .on::<req::GotoDefinition>(handlers::handle_goto_definition)?
 470         .on::<req::GotoImplementation>(handlers::handle_goto_implementation)?
 471         .on::<req::GotoTypeDefinition>(handlers::handle_goto_type_definition)?
 472         .on::<req::ParentModule>(handlers::handle_parent_module)?
 473         .on::<req::Runnables>(handlers::handle_runnables)?
 474         .on::<req::DecorationsRequest>(handlers::handle_decorations)?
 475         .on::<req::Completion>(handlers::handle_completion)?
 476         .on::<req::CodeActionRequest>(handlers::handle_code_action)?
 477         .on::<req::CodeLensRequest>(handlers::handle_code_lens)?
 478         .on::<req::CodeLensResolve>(handlers::handle_code_lens_resolve)?
 479         .on::<req::FoldingRangeRequest>(handlers::handle_folding_range)?
 480         .on::<req::SignatureHelpRequest>(handlers::handle_signature_help)?
 481         .on::<req::HoverRequest>(handlers::handle_hover)?
 482         .on::<req::PrepareRenameRequest>(handlers::handle_prepare_rename)?
 483         .on::<req::Rename>(handlers::handle_rename)?
 484         .on::<req::References>(handlers::handle_references)?
 485         .on::<req::Formatting>(handlers::handle_formatting)?
 486         .on::<req::DocumentHighlightRequest>(handlers::handle_document_highlight)?
 487         .on::<req::InlayHints>(handlers::handle_inlay_hints)?
 488         .finish();
 489     Ok(())
 490 }
 491
 492 fn on_notification(
 493     msg_sender: &Sender<Message>,
 494     state: &mut WorldState,
 495     pending_requests: &mut PendingRequests,
 496     subs: &mut Subscriptions,
 497     not: Notification,
 498 ) -> Result<()> {
 499     let not = match notification_cast::<req::Cancel>(not) {
 500         Ok(params) => {
 501             let id: RequestId = match params.id {
 502                 NumberOrString::Number(id) => id.into(),
 503                 NumberOrString::String(id) => id.into(),
 504             };
 505             if pending_requests.cancel(&id) {
 506                 let response = Response::new_err(
 507                     id,
 508                     ErrorCode::RequestCanceled as i32,
 509                     "canceled by client".to_string(),
 510                 );
 511                 msg_sender.send(response.into()).unwrap()
 512             }
 513             return Ok(());
 514         }
 515         Err(not) => not,
 516     };
 517     let not = match notification_cast::<req::DidOpenTextDocument>(not) {
 518         Ok(params) => {
 519             let uri = params.text_document.uri;
 520             let path = uri.to_file_path().map_err(|()| format!("invalid uri: {}", uri))?;
 521             if let Some(file_id) =
 522                 state.vfs.write().add_file_overlay(&path, params.text_document.text)
 523             {
 524                 subs.add_sub(FileId(file_id.0));
 525             }
 526             return Ok(());
 527         }
 528         Err(not) => not,
 529     };
 530     let not = match notification_cast::<req::DidChangeTextDocument>(not) {
 531         Ok(mut params) => {
 532             let uri = params.text_document.uri;
 533             let path = uri.to_file_path().map_err(|()| format!("invalid uri: {}", uri))?;
 534             let text =
 535                 params.content_changes.pop().ok_or_else(|| "empty changes".to_string())?.text;
 536             state.vfs.write().change_file_overlay(path.as_path(), text);
 537             return Ok(());
 538         }
 539         Err(not) => not,
 540     };
 541     let not = match notification_cast::<req::DidSaveTextDocument>(not) {
 542         Ok(_params) => {
 543             state.check_watcher.update();
 544             return Ok(());
 545         }
 546         Err(not) => not,
 547     };
 548     let not = match notification_cast::<req::DidCloseTextDocument>(not) {
 549         Ok(params) => {
 550             let uri = params.text_document.uri;
 551             let path = uri.to_file_path().map_err(|()| format!("invalid uri: {}", uri))?;
 552             if let Some(file_id) = state.vfs.write().remove_file_overlay(path.as_path()) {
 553                 subs.remove_sub(FileId(file_id.0));
 554             }
 555             let params =
 556                 req::PublishDiagnosticsParams { uri, diagnostics: Vec::new(), version: None };
 557             let not = notification_new::<req::PublishDiagnostics>(params);
 558             msg_sender.send(not.into()).unwrap();
 559             return Ok(());
 560         }
 561         Err(not) => not,
 562     };
 563     let not = match notification_cast::<req::DidChangeConfiguration>(not) {
 564         Ok(_params) => {
 565             return Ok(());
 566         }
 567         Err(not) => not,
 568     };
 569     let not = match notification_cast::<req::DidChangeWatchedFiles>(not) {
 570         Ok(params) => {
 571             let mut vfs = state.vfs.write();
 572             for change in params.changes {
 573                 let uri = change.uri;
 574                 let path = uri.to_file_path().map_err(|()| format!("invalid uri: {}", uri))?;
 575                 vfs.notify_changed(path)
 576             }
 577             return Ok(());
 578         }
 579         Err(not) => not,
 580     };
 581     log::error!("unhandled notification: {:?}", not);
 582     Ok(())
 583 }
 584
 585 struct PoolDispatcher<'a> {
 586     req: Option<Request>,
 587     pool: &'a ThreadPool,
 588     world: &'a mut WorldState,
 589     pending_requests: &'a mut PendingRequests,
 590     msg_sender: &'a Sender<Message>,
 591     sender: &'a Sender<Task>,
 592     request_received: Instant,
 593 }
 594
 595 impl<'a> PoolDispatcher<'a> {
 596     /// Dispatches the request onto the current thread
 597     fn on_sync<R>(
 598         &mut self,
 599         f: fn(&mut WorldState, R::Params) -> Result<R::Result>,
 600     ) -> Result<&mut Self>
 601     where
 602         R: req::Request + 'static,
 603         R::Params: DeserializeOwned + panic::UnwindSafe + 'static,
 604         R::Result: Serialize + 'static,
 605     {
 606         let (id, params) = match self.parse::<R>() {
 607             Some(it) => it,
 608             None => {
 609                 return Ok(self);
 610             }
 611         };
 612         let world = panic::AssertUnwindSafe(&mut *self.world);
 613         let task = panic::catch_unwind(move || {
 614             let result = f(world.0, params);
 615             result_to_task::<R>(id, result)
 616         })
 617         .map_err(|_| format!("sync task {:?} panicked", R::METHOD))?;
 618         on_task(task, self.msg_sender, self.pending_requests, self.world);
 619         Ok(self)
 620     }
 621
 622     /// Dispatches the request onto thread pool
 623     fn on<R>(&mut self, f: fn(WorldSnapshot, R::Params) -> Result<R::Result>) -> Result<&mut Self>
 624     where
 625         R: req::Request + 'static,
 626         R::Params: DeserializeOwned + Send + 'static,
 627         R::Result: Serialize + 'static,
 628     {
 629         let (id, params) = match self.parse::<R>() {
 630             Some(it) => it,
 631             None => {
 632                 return Ok(self);
 633             }
 634         };
 635
 636         self.pool.execute({
 637             let world = self.world.snapshot();
 638             let sender = self.sender.clone();
 639             move || {
 640                 let result = f(world, params);
 641                 let task = result_to_task::<R>(id, result);
 642                 sender.send(task).unwrap();
 643             }
 644         });
 645
 646         Ok(self)
 647     }
 648
 649     fn parse<R>(&mut self) -> Option<(RequestId, R::Params)>
 650     where
 651         R: req::Request + 'static,
 652         R::Params: DeserializeOwned + 'static,
 653     {
 654         let req = self.req.take()?;
 655         let (id, params) = match req.extract::<R::Params>(R::METHOD) {
 656             Ok(it) => it,
 657             Err(req) => {
 658                 self.req = Some(req);
 659                 return None;
 660             }
 661         };
 662         self.pending_requests.start(PendingRequest {
 663             id: id.clone(),
 664             method: R::METHOD.to_string(),
 665             received: self.request_received,
 666         });
 667         Some((id, params))
 668     }
 669
 670     fn finish(&mut self) {
 671         match self.req.take() {
 672             None => (),
 673             Some(req) => {
 674                 log::error!("unknown request: {:?}", req);
 675                 let resp = Response::new_err(
 676                     req.id,
 677                     ErrorCode::MethodNotFound as i32,
 678                     "unknown request".to_string(),
 679                 );
 680                 self.msg_sender.send(resp.into()).unwrap();
 681             }
 682         }
 683     }
 684 }
 685
 686 fn result_to_task<R>(id: RequestId, result: Result<R::Result>) -> Task
 687 where
 688     R: req::Request + 'static,
 689     R::Params: DeserializeOwned + 'static,
 690     R::Result: Serialize + 'static,
 691 {
 692     let response = match result {
 693         Ok(resp) => Response::new_ok(id, &resp),
 694         Err(e) => match e.downcast::<LspError>() {
 695             Ok(lsp_error) => Response::new_err(id, lsp_error.code, lsp_error.message),
 696             Err(e) => {
 697                 if is_canceled(&e) {
 698                     // FIXME: When https://github.com/Microsoft/vscode-languageserver-node/issues/457
 699                     // gets fixed, we can return the proper response.
 700                     // This works around the issue where "content modified" error would continuously
 701                     // show an message pop-up in VsCode
 702                     // Response::err(
 703                     //     id,
 704                     //     ErrorCode::ContentModified as i32,
 705                     //     "content modified".to_string(),
 706                     // )
 707                     Response::new_ok(id, ())
 708                 } else {
 709                     Response::new_err(id, ErrorCode::InternalError as i32, e.to_string())
 710                 }
 711             }
 712         },
 713     };
 714     Task::Respond(response)
 715 }
 716
 717 fn update_file_notifications_on_threadpool(
 718     pool: &ThreadPool,
 719     world: WorldSnapshot,
 720     publish_decorations: bool,
 721     sender: Sender<Task>,
 722     subscriptions: Vec<FileId>,
 723 ) {
 724     log::trace!("updating notifications for {:?}", subscriptions);
 725     let publish_diagnostics = world.feature_flags().get("lsp.diagnostics");
 726     pool.execute(move || {
 727         for file_id in subscriptions {
 728             if publish_diagnostics {
 729                 match handlers::publish_diagnostics(&world, file_id) {
 730                     Err(e) => {
 731                         if !is_canceled(&e) {
 732                             log::error!("failed to compute diagnostics: {:?}", e);
 733                         }
 734                     }
 735                     Ok(params) => {
 736                         let not = notification_new::<req::PublishDiagnostics>(params);
 737                         sender.send(Task::Notify(not)).unwrap();
 738                     }
 739                 }
 740             }
 741             if publish_decorations {
 742                 match handlers::publish_decorations(&world, file_id) {
 743                     Err(e) => {
 744                         if !is_canceled(&e) {
 745                             log::error!("failed to compute decorations: {:?}", e);
 746                         }
 747                     }
 748                     Ok(params) => {
 749                         let not = notification_new::<req::PublishDecorations>(params);
 750                         sender.send(Task::Notify(not)).unwrap();
 751                     }
 752                 }
 753             }
 754         }
 755     });
 756 }
 757
 758 pub fn show_message(typ: req::MessageType, message: impl Into<String>, sender: &Sender<Message>) {
 759     let message = message.into();
 760     let params = req::ShowMessageParams { typ, message };
 761     let not = notification_new::<req::ShowMessage>(params);
 762     sender.send(not.into()).unwrap();
 763 }
 764
 765 fn is_canceled(e: &Box<dyn std::error::Error + Send + Sync>) -> bool {
 766     e.downcast_ref::<Canceled>().is_some()
 767 }
 768
 769 fn notification_is<N: lsp_types::notification::Notification>(notification: &Notification) -> bool {
 770     notification.method == N::METHOD
 771 }
 772
 773 fn notification_cast<N>(notification: Notification) -> std::result::Result<N::Params, Notification>
 774 where
 775     N: lsp_types::notification::Notification,
 776     N::Params: DeserializeOwned,
 777 {
 778     notification.extract(N::METHOD)
 779 }
 780
 781 fn notification_new<N>(params: N::Params) -> Notification
 782 where
 783     N: lsp_types::notification::Notification,
 784     N::Params: Serialize,
 785 {
 786     Notification::new(N::METHOD.to_string(), params)
 787 }
 788
 789 fn request_new<R>(id: RequestId, params: R::Params) -> Request
 790 where
 791     R: lsp_types::request::Request,
 792     R::Params: Serialize,
 793 {
 794     Request::new(id, R::METHOD.to_string(), params)
 795 }