lsp.rs

   1mod input_handler;
   2
   3pub use lsp_types::request::*;
   4pub use lsp_types::*;
   5
   6use anyhow::{anyhow, Context, Result};
   7use collections::HashMap;
   8use futures::{channel::oneshot, io::BufWriter, select, AsyncRead, AsyncWrite, Future, FutureExt};
   9use gpui::{AppContext, AsyncAppContext, BackgroundExecutor, SharedString, Task};
  10use notification::DidChangeWorkspaceFolders;
  11use parking_lot::{Mutex, RwLock};
  12use postage::{barrier, prelude::Stream};
  13use schemars::{
  14    gen::SchemaGenerator,
  15    schema::{InstanceType, Schema, SchemaObject},
  16    JsonSchema,
  17};
  18use serde::{de::DeserializeOwned, Deserialize, Serialize};
  19use serde_json::{json, value::RawValue, Value};
  20use smol::{
  21    channel,
  22    io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
  23    process::Child,
  24};
  25use text::BufferId;
  26
  27use std::{
  28    collections::BTreeSet,
  29    ffi::{OsStr, OsString},
  30    fmt,
  31    io::Write,
  32    ops::{Deref, DerefMut},
  33    path::PathBuf,
  34    pin::Pin,
  35    sync::{
  36        atomic::{AtomicI32, Ordering::SeqCst},
  37        Arc, Weak,
  38    },
  39    task::Poll,
  40    time::{Duration, Instant},
  41};
  42use std::{path::Path, process::Stdio};
  43use util::{ResultExt, TryFutureExt};
  44
  45const JSON_RPC_VERSION: &str = "2.0";
  46const CONTENT_LEN_HEADER: &str = "Content-Length: ";
  47
  48const LSP_REQUEST_TIMEOUT: Duration = Duration::from_secs(60 * 2);
  49const SERVER_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5);
  50
  51type NotificationHandler = Box<dyn Send + FnMut(Option<RequestId>, Value, AsyncAppContext)>;
  52type ResponseHandler = Box<dyn Send + FnOnce(Result<String, Error>)>;
  53type IoHandler = Box<dyn Send + FnMut(IoKind, &str)>;
  54
  55/// Kind of language server stdio given to an IO handler.
  56#[derive(Debug, Clone, Copy)]
  57pub enum IoKind {
  58    StdOut,
  59    StdIn,
  60    StdErr,
  61}
  62
  63/// Represents a launchable language server. This can either be a standalone binary or the path
  64/// to a runtime with arguments to instruct it to launch the actual language server file.
  65#[derive(Debug, Clone, Deserialize)]
  66pub struct LanguageServerBinary {
  67    pub path: PathBuf,
  68    pub arguments: Vec<OsString>,
  69    pub env: Option<HashMap<String, String>>,
  70}
  71
  72/// Configures the search (and installation) of language servers.
  73#[derive(Debug, Clone, Deserialize)]
  74pub struct LanguageServerBinaryOptions {
  75    /// Whether the adapter should look at the users system
  76    pub allow_path_lookup: bool,
  77    /// Whether the adapter should download its own version
  78    pub allow_binary_download: bool,
  79}
  80
  81/// A running language server process.
  82pub struct LanguageServer {
  83    server_id: LanguageServerId,
  84    next_id: AtomicI32,
  85    outbound_tx: channel::Sender<String>,
  86    name: LanguageServerName,
  87    process_name: Arc<str>,
  88    binary: LanguageServerBinary,
  89    capabilities: RwLock<ServerCapabilities>,
  90    /// Configuration sent to the server, stored for display in the language server logs
  91    /// buffer. This is represented as the message sent to the LSP in order to avoid cloning it (can
  92    /// be large in cases like sending schemas to the json server).
  93    configuration: Arc<DidChangeConfigurationParams>,
  94    code_action_kinds: Option<Vec<CodeActionKind>>,
  95    notification_handlers: Arc<Mutex<HashMap<&'static str, NotificationHandler>>>,
  96    response_handlers: Arc<Mutex<Option<HashMap<RequestId, ResponseHandler>>>>,
  97    io_handlers: Arc<Mutex<HashMap<i32, IoHandler>>>,
  98    executor: BackgroundExecutor,
  99    #[allow(clippy::type_complexity)]
 100    io_tasks: Mutex<Option<(Task<Option<()>>, Task<Option<()>>)>>,
 101    output_done_rx: Mutex<Option<barrier::Receiver>>,
 102    server: Arc<Mutex<Option<Child>>>,
 103    workspace_folders: Arc<Mutex<BTreeSet<Url>>>,
 104    registered_buffers: Arc<Mutex<HashMap<BufferId, Url>>>,
 105}
 106
 107/// Identifies a running language server.
 108#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
 109#[repr(transparent)]
 110pub struct LanguageServerId(pub usize);
 111
 112impl LanguageServerId {
 113    pub fn from_proto(id: u64) -> Self {
 114        Self(id as usize)
 115    }
 116
 117    pub fn to_proto(self) -> u64 {
 118        self.0 as u64
 119    }
 120}
 121
 122/// A name of a language server.
 123#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Deserialize, Serialize)]
 124pub struct LanguageServerName(pub SharedString);
 125
 126impl std::fmt::Display for LanguageServerName {
 127    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
 128        std::fmt::Display::fmt(&self.0, f)
 129    }
 130}
 131
 132impl AsRef<str> for LanguageServerName {
 133    fn as_ref(&self) -> &str {
 134        self.0.as_ref()
 135    }
 136}
 137
 138impl AsRef<OsStr> for LanguageServerName {
 139    fn as_ref(&self) -> &OsStr {
 140        self.0.as_ref().as_ref()
 141    }
 142}
 143
 144impl JsonSchema for LanguageServerName {
 145    fn schema_name() -> String {
 146        "LanguageServerName".into()
 147    }
 148
 149    fn json_schema(_: &mut SchemaGenerator) -> Schema {
 150        SchemaObject {
 151            instance_type: Some(InstanceType::String.into()),
 152            ..Default::default()
 153        }
 154        .into()
 155    }
 156}
 157
 158impl LanguageServerName {
 159    pub const fn new_static(s: &'static str) -> Self {
 160        Self(SharedString::new_static(s))
 161    }
 162
 163    pub fn from_proto(s: String) -> Self {
 164        Self(s.into())
 165    }
 166}
 167
 168impl<'a> From<&'a str> for LanguageServerName {
 169    fn from(str: &'a str) -> LanguageServerName {
 170        LanguageServerName(str.to_string().into())
 171    }
 172}
 173
 174/// Handle to a language server RPC activity subscription.
 175pub enum Subscription {
 176    Notification {
 177        method: &'static str,
 178        notification_handlers: Option<Arc<Mutex<HashMap<&'static str, NotificationHandler>>>>,
 179    },
 180    Io {
 181        id: i32,
 182        io_handlers: Option<Weak<Mutex<HashMap<i32, IoHandler>>>>,
 183    },
 184}
 185
 186/// Language server protocol RPC request message ID.
 187///
 188/// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#requestMessage)
 189#[derive(Debug, Clone, Eq, PartialEq, Hash, Serialize, Deserialize)]
 190#[serde(untagged)]
 191pub enum RequestId {
 192    Int(i32),
 193    Str(String),
 194}
 195
 196/// Language server protocol RPC request message.
 197///
 198/// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#requestMessage)
 199#[derive(Serialize, Deserialize)]
 200pub struct Request<'a, T> {
 201    jsonrpc: &'static str,
 202    id: RequestId,
 203    method: &'a str,
 204    params: T,
 205}
 206
 207/// Language server protocol RPC request response message before it is deserialized into a concrete type.
 208#[derive(Serialize, Deserialize)]
 209struct AnyResponse<'a> {
 210    jsonrpc: &'a str,
 211    id: RequestId,
 212    #[serde(default)]
 213    error: Option<Error>,
 214    #[serde(borrow)]
 215    result: Option<&'a RawValue>,
 216}
 217
 218/// Language server protocol RPC request response message.
 219///
 220/// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#responseMessage)
 221#[derive(Serialize)]
 222struct Response<T> {
 223    jsonrpc: &'static str,
 224    id: RequestId,
 225    #[serde(flatten)]
 226    value: LspResult<T>,
 227}
 228
 229#[derive(Serialize)]
 230#[serde(rename_all = "snake_case")]
 231enum LspResult<T> {
 232    #[serde(rename = "result")]
 233    Ok(Option<T>),
 234    Error(Option<Error>),
 235}
 236
 237/// Language server protocol RPC notification message.
 238///
 239/// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#notificationMessage)
 240#[derive(Serialize, Deserialize)]
 241struct Notification<'a, T> {
 242    jsonrpc: &'static str,
 243    #[serde(borrow)]
 244    method: &'a str,
 245    params: T,
 246}
 247
 248/// Language server RPC notification message before it is deserialized into a concrete type.
 249#[derive(Debug, Clone, Deserialize)]
 250struct AnyNotification {
 251    #[serde(default)]
 252    id: Option<RequestId>,
 253    method: String,
 254    #[serde(default)]
 255    params: Option<Value>,
 256}
 257
 258#[derive(Debug, Serialize, Deserialize)]
 259struct Error {
 260    message: String,
 261}
 262
 263pub trait LspRequestFuture<O>: Future<Output = O> {
 264    fn id(&self) -> i32;
 265}
 266
 267struct LspRequest<F> {
 268    id: i32,
 269    request: F,
 270}
 271
 272impl<F> LspRequest<F> {
 273    pub fn new(id: i32, request: F) -> Self {
 274        Self { id, request }
 275    }
 276}
 277
 278impl<F: Future> Future for LspRequest<F> {
 279    type Output = F::Output;
 280
 281    fn poll(self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
 282        // SAFETY: This is standard pin projection, we're pinned so our fields must be pinned.
 283        let inner = unsafe { Pin::new_unchecked(&mut self.get_unchecked_mut().request) };
 284        inner.poll(cx)
 285    }
 286}
 287
 288impl<F: Future> LspRequestFuture<F::Output> for LspRequest<F> {
 289    fn id(&self) -> i32 {
 290        self.id
 291    }
 292}
 293
 294/// Combined capabilities of the server and the adapter.
 295#[derive(Debug)]
 296pub struct AdapterServerCapabilities {
 297    // Reported capabilities by the server
 298    pub server_capabilities: ServerCapabilities,
 299    // List of code actions supported by the LspAdapter matching the server
 300    pub code_action_kinds: Option<Vec<CodeActionKind>>,
 301}
 302
 303/// Experimental: Informs the end user about the state of the server
 304///
 305/// [Rust Analyzer Specification](https://github.com/rust-lang/rust-analyzer/blob/master/docs/dev/lsp-extensions.md#server-status)
 306#[derive(Debug)]
 307pub enum ServerStatus {}
 308
 309/// Other(String) variant to handle unknown values due to this still being experimental
 310#[derive(Debug, PartialEq, Deserialize, Serialize, Clone)]
 311#[serde(rename_all = "camelCase")]
 312pub enum ServerHealthStatus {
 313    Ok,
 314    Warning,
 315    Error,
 316    Other(String),
 317}
 318
 319#[derive(Debug, PartialEq, Deserialize, Serialize, Clone)]
 320#[serde(rename_all = "camelCase")]
 321pub struct ServerStatusParams {
 322    pub health: ServerHealthStatus,
 323    pub message: Option<String>,
 324}
 325
 326impl lsp_types::notification::Notification for ServerStatus {
 327    type Params = ServerStatusParams;
 328    const METHOD: &'static str = "experimental/serverStatus";
 329}
 330
 331impl LanguageServer {
 332    /// Starts a language server process.
 333    pub fn new(
 334        stderr_capture: Arc<Mutex<Option<String>>>,
 335        server_id: LanguageServerId,
 336        server_name: LanguageServerName,
 337        binary: LanguageServerBinary,
 338        root_path: &Path,
 339        code_action_kinds: Option<Vec<CodeActionKind>>,
 340        cx: AsyncAppContext,
 341    ) -> Result<Self> {
 342        let working_dir = if root_path.is_dir() {
 343            root_path
 344        } else {
 345            root_path.parent().unwrap_or_else(|| Path::new("/"))
 346        };
 347
 348        log::info!(
 349            "starting language server process. binary path: {:?}, working directory: {:?}, args: {:?}",
 350            binary.path,
 351            working_dir,
 352            &binary.arguments
 353        );
 354
 355        let mut server = util::command::new_smol_command(&binary.path)
 356            .current_dir(working_dir)
 357            .args(&binary.arguments)
 358            .envs(binary.env.clone().unwrap_or_default())
 359            .stdin(Stdio::piped())
 360            .stdout(Stdio::piped())
 361            .stderr(Stdio::piped())
 362            .kill_on_drop(true)
 363            .spawn()
 364            .with_context(|| {
 365                format!(
 366                    "failed to spawn command. path: {:?}, working directory: {:?}, args: {:?}",
 367                    binary.path, working_dir, &binary.arguments
 368                )
 369            })?;
 370
 371        let stdin = server.stdin.take().unwrap();
 372        let stdout = server.stdout.take().unwrap();
 373        let stderr = server.stderr.take().unwrap();
 374        let server = Self::new_internal(
 375            server_id,
 376            server_name,
 377            stdin,
 378            stdout,
 379            Some(stderr),
 380            stderr_capture,
 381            Some(server),
 382            code_action_kinds,
 383            binary,
 384            cx,
 385            move |notification| {
 386                log::info!(
 387                    "Language server with id {} sent unhandled notification {}:\n{}",
 388                    server_id,
 389                    notification.method,
 390                    serde_json::to_string_pretty(&notification.params).unwrap(),
 391                );
 392            },
 393        );
 394
 395        Ok(server)
 396    }
 397
 398    #[allow(clippy::too_many_arguments)]
 399    fn new_internal<Stdin, Stdout, Stderr, F>(
 400        server_id: LanguageServerId,
 401        server_name: LanguageServerName,
 402        stdin: Stdin,
 403        stdout: Stdout,
 404        stderr: Option<Stderr>,
 405        stderr_capture: Arc<Mutex<Option<String>>>,
 406        server: Option<Child>,
 407        code_action_kinds: Option<Vec<CodeActionKind>>,
 408        binary: LanguageServerBinary,
 409        cx: AsyncAppContext,
 410        on_unhandled_notification: F,
 411    ) -> Self
 412    where
 413        Stdin: AsyncWrite + Unpin + Send + 'static,
 414        Stdout: AsyncRead + Unpin + Send + 'static,
 415        Stderr: AsyncRead + Unpin + Send + 'static,
 416        F: FnMut(AnyNotification) + 'static + Send + Sync + Clone,
 417    {
 418        let (outbound_tx, outbound_rx) = channel::unbounded::<String>();
 419        let (output_done_tx, output_done_rx) = barrier::channel();
 420        let notification_handlers =
 421            Arc::new(Mutex::new(HashMap::<_, NotificationHandler>::default()));
 422        let response_handlers =
 423            Arc::new(Mutex::new(Some(HashMap::<_, ResponseHandler>::default())));
 424        let io_handlers = Arc::new(Mutex::new(HashMap::default()));
 425
 426        let stdout_input_task = cx.spawn({
 427            let on_unhandled_notification = on_unhandled_notification.clone();
 428            let notification_handlers = notification_handlers.clone();
 429            let response_handlers = response_handlers.clone();
 430            let io_handlers = io_handlers.clone();
 431            move |cx| {
 432                Self::handle_input(
 433                    stdout,
 434                    on_unhandled_notification,
 435                    notification_handlers,
 436                    response_handlers,
 437                    io_handlers,
 438                    cx,
 439                )
 440                .log_err()
 441            }
 442        });
 443        let stderr_input_task = stderr
 444            .map(|stderr| {
 445                let io_handlers = io_handlers.clone();
 446                let stderr_captures = stderr_capture.clone();
 447                cx.spawn(|_| Self::handle_stderr(stderr, io_handlers, stderr_captures).log_err())
 448            })
 449            .unwrap_or_else(|| Task::ready(None));
 450        let input_task = cx.spawn(|_| async move {
 451            let (stdout, stderr) = futures::join!(stdout_input_task, stderr_input_task);
 452            stdout.or(stderr)
 453        });
 454        let output_task = cx.background_executor().spawn({
 455            Self::handle_output(
 456                stdin,
 457                outbound_rx,
 458                output_done_tx,
 459                response_handlers.clone(),
 460                io_handlers.clone(),
 461            )
 462            .log_err()
 463        });
 464
 465        let configuration = DidChangeConfigurationParams {
 466            settings: Value::Null,
 467        }
 468        .into();
 469
 470        Self {
 471            server_id,
 472            notification_handlers,
 473            response_handlers,
 474            io_handlers,
 475            name: server_name,
 476            process_name: binary
 477                .path
 478                .file_name()
 479                .map(|name| Arc::from(name.to_string_lossy()))
 480                .unwrap_or_default(),
 481            binary,
 482            capabilities: Default::default(),
 483            configuration,
 484            code_action_kinds,
 485            next_id: Default::default(),
 486            outbound_tx,
 487            executor: cx.background_executor().clone(),
 488            io_tasks: Mutex::new(Some((input_task, output_task))),
 489            output_done_rx: Mutex::new(Some(output_done_rx)),
 490            server: Arc::new(Mutex::new(server)),
 491            workspace_folders: Default::default(),
 492            registered_buffers: Default::default(),
 493        }
 494    }
 495
 496    /// List of code action kinds this language server reports being able to emit.
 497    pub fn code_action_kinds(&self) -> Option<Vec<CodeActionKind>> {
 498        self.code_action_kinds.clone()
 499    }
 500
 501    async fn handle_input<Stdout, F>(
 502        stdout: Stdout,
 503        mut on_unhandled_notification: F,
 504        notification_handlers: Arc<Mutex<HashMap<&'static str, NotificationHandler>>>,
 505        response_handlers: Arc<Mutex<Option<HashMap<RequestId, ResponseHandler>>>>,
 506        io_handlers: Arc<Mutex<HashMap<i32, IoHandler>>>,
 507        cx: AsyncAppContext,
 508    ) -> anyhow::Result<()>
 509    where
 510        Stdout: AsyncRead + Unpin + Send + 'static,
 511        F: FnMut(AnyNotification) + 'static + Send,
 512    {
 513        use smol::stream::StreamExt;
 514        let stdout = BufReader::new(stdout);
 515        let _clear_response_handlers = util::defer({
 516            let response_handlers = response_handlers.clone();
 517            move || {
 518                response_handlers.lock().take();
 519            }
 520        });
 521        let mut input_handler = input_handler::LspStdoutHandler::new(
 522            stdout,
 523            response_handlers,
 524            io_handlers,
 525            cx.background_executor().clone(),
 526        );
 527
 528        while let Some(msg) = input_handler.notifications_channel.next().await {
 529            {
 530                let mut notification_handlers = notification_handlers.lock();
 531                if let Some(handler) = notification_handlers.get_mut(msg.method.as_str()) {
 532                    handler(msg.id, msg.params.unwrap_or(Value::Null), cx.clone());
 533                } else {
 534                    drop(notification_handlers);
 535                    on_unhandled_notification(msg);
 536                }
 537            }
 538
 539            // Don't starve the main thread when receiving lots of notifications at once.
 540            smol::future::yield_now().await;
 541        }
 542        input_handler.loop_handle.await
 543    }
 544
 545    async fn handle_stderr<Stderr>(
 546        stderr: Stderr,
 547        io_handlers: Arc<Mutex<HashMap<i32, IoHandler>>>,
 548        stderr_capture: Arc<Mutex<Option<String>>>,
 549    ) -> anyhow::Result<()>
 550    where
 551        Stderr: AsyncRead + Unpin + Send + 'static,
 552    {
 553        let mut stderr = BufReader::new(stderr);
 554        let mut buffer = Vec::new();
 555
 556        loop {
 557            buffer.clear();
 558
 559            let bytes_read = stderr.read_until(b'\n', &mut buffer).await?;
 560            if bytes_read == 0 {
 561                return Ok(());
 562            }
 563
 564            if let Ok(message) = std::str::from_utf8(&buffer) {
 565                log::trace!("incoming stderr message:{message}");
 566                for handler in io_handlers.lock().values_mut() {
 567                    handler(IoKind::StdErr, message);
 568                }
 569
 570                if let Some(stderr) = stderr_capture.lock().as_mut() {
 571                    stderr.push_str(message);
 572                }
 573            }
 574
 575            // Don't starve the main thread when receiving lots of messages at once.
 576            smol::future::yield_now().await;
 577        }
 578    }
 579
 580    async fn handle_output<Stdin>(
 581        stdin: Stdin,
 582        outbound_rx: channel::Receiver<String>,
 583        output_done_tx: barrier::Sender,
 584        response_handlers: Arc<Mutex<Option<HashMap<RequestId, ResponseHandler>>>>,
 585        io_handlers: Arc<Mutex<HashMap<i32, IoHandler>>>,
 586    ) -> anyhow::Result<()>
 587    where
 588        Stdin: AsyncWrite + Unpin + Send + 'static,
 589    {
 590        let mut stdin = BufWriter::new(stdin);
 591        let _clear_response_handlers = util::defer({
 592            let response_handlers = response_handlers.clone();
 593            move || {
 594                response_handlers.lock().take();
 595            }
 596        });
 597        let mut content_len_buffer = Vec::new();
 598        while let Ok(message) = outbound_rx.recv().await {
 599            log::trace!("outgoing message:{}", message);
 600            for handler in io_handlers.lock().values_mut() {
 601                handler(IoKind::StdIn, &message);
 602            }
 603
 604            content_len_buffer.clear();
 605            write!(content_len_buffer, "{}", message.len()).unwrap();
 606            stdin.write_all(CONTENT_LEN_HEADER.as_bytes()).await?;
 607            stdin.write_all(&content_len_buffer).await?;
 608            stdin.write_all("\r\n\r\n".as_bytes()).await?;
 609            stdin.write_all(message.as_bytes()).await?;
 610            stdin.flush().await?;
 611        }
 612        drop(output_done_tx);
 613        Ok(())
 614    }
 615
 616    pub fn default_initialize_params(&self, cx: &AppContext) -> InitializeParams {
 617        #[allow(deprecated)]
 618        InitializeParams {
 619            process_id: None,
 620            root_path: None,
 621            root_uri: None,
 622            initialization_options: None,
 623            capabilities: ClientCapabilities {
 624                general: Some(GeneralClientCapabilities {
 625                    position_encodings: Some(vec![PositionEncodingKind::UTF16]),
 626                    ..Default::default()
 627                }),
 628                workspace: Some(WorkspaceClientCapabilities {
 629                    configuration: Some(true),
 630                    did_change_watched_files: Some(DidChangeWatchedFilesClientCapabilities {
 631                        dynamic_registration: Some(true),
 632                        relative_pattern_support: Some(true),
 633                    }),
 634                    did_change_configuration: Some(DynamicRegistrationClientCapabilities {
 635                        dynamic_registration: Some(true),
 636                    }),
 637                    workspace_folders: Some(true),
 638                    symbol: Some(WorkspaceSymbolClientCapabilities {
 639                        resolve_support: None,
 640                        ..WorkspaceSymbolClientCapabilities::default()
 641                    }),
 642                    inlay_hint: Some(InlayHintWorkspaceClientCapabilities {
 643                        refresh_support: Some(true),
 644                    }),
 645                    diagnostic: Some(DiagnosticWorkspaceClientCapabilities {
 646                        refresh_support: None,
 647                    }),
 648                    workspace_edit: Some(WorkspaceEditClientCapabilities {
 649                        resource_operations: Some(vec![
 650                            ResourceOperationKind::Create,
 651                            ResourceOperationKind::Rename,
 652                            ResourceOperationKind::Delete,
 653                        ]),
 654                        document_changes: Some(true),
 655                        snippet_edit_support: Some(true),
 656                        ..WorkspaceEditClientCapabilities::default()
 657                    }),
 658                    file_operations: Some(WorkspaceFileOperationsClientCapabilities {
 659                        dynamic_registration: Some(false),
 660                        did_rename: Some(true),
 661                        will_rename: Some(true),
 662                        ..Default::default()
 663                    }),
 664                    apply_edit: Some(true),
 665                    ..Default::default()
 666                }),
 667                text_document: Some(TextDocumentClientCapabilities {
 668                    definition: Some(GotoCapability {
 669                        link_support: Some(true),
 670                        dynamic_registration: None,
 671                    }),
 672                    code_action: Some(CodeActionClientCapabilities {
 673                        code_action_literal_support: Some(CodeActionLiteralSupport {
 674                            code_action_kind: CodeActionKindLiteralSupport {
 675                                value_set: vec![
 676                                    CodeActionKind::REFACTOR.as_str().into(),
 677                                    CodeActionKind::QUICKFIX.as_str().into(),
 678                                    CodeActionKind::SOURCE.as_str().into(),
 679                                ],
 680                            },
 681                        }),
 682                        data_support: Some(true),
 683                        resolve_support: Some(CodeActionCapabilityResolveSupport {
 684                            properties: vec![
 685                                "kind".to_string(),
 686                                "diagnostics".to_string(),
 687                                "isPreferred".to_string(),
 688                                "disabled".to_string(),
 689                                "edit".to_string(),
 690                                "command".to_string(),
 691                            ],
 692                        }),
 693                        ..Default::default()
 694                    }),
 695                    completion: Some(CompletionClientCapabilities {
 696                        completion_item: Some(CompletionItemCapability {
 697                            snippet_support: Some(true),
 698                            resolve_support: Some(CompletionItemCapabilityResolveSupport {
 699                                properties: vec![
 700                                    "additionalTextEdits".to_string(),
 701                                    "command".to_string(),
 702                                    "documentation".to_string(),
 703                                    // NB: Do not have this resolved, otherwise Zed becomes slow to complete things
 704                                    // "textEdit".to_string(),
 705                                ],
 706                            }),
 707                            insert_replace_support: Some(true),
 708                            label_details_support: Some(true),
 709                            ..Default::default()
 710                        }),
 711                        completion_list: Some(CompletionListCapability {
 712                            item_defaults: Some(vec![
 713                                "commitCharacters".to_owned(),
 714                                "editRange".to_owned(),
 715                                "insertTextMode".to_owned(),
 716                                "insertTextFormat".to_owned(),
 717                                "data".to_owned(),
 718                            ]),
 719                        }),
 720                        context_support: Some(true),
 721                        ..Default::default()
 722                    }),
 723                    rename: Some(RenameClientCapabilities {
 724                        prepare_support: Some(true),
 725                        ..Default::default()
 726                    }),
 727                    hover: Some(HoverClientCapabilities {
 728                        content_format: Some(vec![MarkupKind::Markdown]),
 729                        dynamic_registration: None,
 730                    }),
 731                    inlay_hint: Some(InlayHintClientCapabilities {
 732                        resolve_support: Some(InlayHintResolveClientCapabilities {
 733                            properties: vec![
 734                                "textEdits".to_string(),
 735                                "tooltip".to_string(),
 736                                "label.tooltip".to_string(),
 737                                "label.location".to_string(),
 738                                "label.command".to_string(),
 739                            ],
 740                        }),
 741                        dynamic_registration: Some(false),
 742                    }),
 743                    publish_diagnostics: Some(PublishDiagnosticsClientCapabilities {
 744                        related_information: Some(true),
 745                        ..Default::default()
 746                    }),
 747                    formatting: Some(DynamicRegistrationClientCapabilities {
 748                        dynamic_registration: Some(true),
 749                    }),
 750                    range_formatting: Some(DynamicRegistrationClientCapabilities {
 751                        dynamic_registration: Some(true),
 752                    }),
 753                    on_type_formatting: Some(DynamicRegistrationClientCapabilities {
 754                        dynamic_registration: Some(true),
 755                    }),
 756                    signature_help: Some(SignatureHelpClientCapabilities {
 757                        signature_information: Some(SignatureInformationSettings {
 758                            documentation_format: Some(vec![
 759                                MarkupKind::Markdown,
 760                                MarkupKind::PlainText,
 761                            ]),
 762                            parameter_information: Some(ParameterInformationSettings {
 763                                label_offset_support: Some(true),
 764                            }),
 765                            active_parameter_support: Some(true),
 766                        }),
 767                        ..SignatureHelpClientCapabilities::default()
 768                    }),
 769                    synchronization: Some(TextDocumentSyncClientCapabilities {
 770                        did_save: Some(true),
 771                        ..TextDocumentSyncClientCapabilities::default()
 772                    }),
 773                    ..TextDocumentClientCapabilities::default()
 774                }),
 775                experimental: Some(json!({
 776                    "serverStatusNotification": true,
 777                    "localDocs": true,
 778                })),
 779                window: Some(WindowClientCapabilities {
 780                    work_done_progress: Some(true),
 781                    show_message: Some(ShowMessageRequestClientCapabilities {
 782                        message_action_item: None,
 783                    }),
 784                    ..Default::default()
 785                }),
 786            },
 787            trace: None,
 788            workspace_folders: None,
 789            client_info: release_channel::ReleaseChannel::try_global(cx).map(|release_channel| {
 790                ClientInfo {
 791                    name: release_channel.display_name().to_string(),
 792                    version: Some(release_channel::AppVersion::global(cx).to_string()),
 793                }
 794            }),
 795            locale: None,
 796
 797            ..Default::default()
 798        }
 799    }
 800
 801    /// Initializes a language server by sending the `Initialize` request.
 802    /// Note that `options` is used directly to construct [`InitializeParams`], which is why it is owned.
 803    ///
 804    /// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#initialize)
 805    pub fn initialize(
 806        mut self,
 807        params: InitializeParams,
 808        configuration: Arc<DidChangeConfigurationParams>,
 809        cx: &AppContext,
 810    ) -> Task<Result<Arc<Self>>> {
 811        cx.spawn(|_| async move {
 812            let response = self.request::<request::Initialize>(params).await?;
 813            if let Some(info) = response.server_info {
 814                self.process_name = info.name.into();
 815            }
 816            self.capabilities = RwLock::new(response.capabilities);
 817            self.configuration = configuration;
 818
 819            self.notify::<notification::Initialized>(&InitializedParams {})?;
 820            Ok(Arc::new(self))
 821        })
 822    }
 823
 824    /// Sends a shutdown request to the language server process and prepares the [`LanguageServer`] to be dropped.
 825    pub fn shutdown(&self) -> Option<impl 'static + Send + Future<Output = Option<()>>> {
 826        if let Some(tasks) = self.io_tasks.lock().take() {
 827            let response_handlers = self.response_handlers.clone();
 828            let next_id = AtomicI32::new(self.next_id.load(SeqCst));
 829            let outbound_tx = self.outbound_tx.clone();
 830            let executor = self.executor.clone();
 831            let mut output_done = self.output_done_rx.lock().take().unwrap();
 832            let shutdown_request = Self::request_internal::<request::Shutdown>(
 833                &next_id,
 834                &response_handlers,
 835                &outbound_tx,
 836                &executor,
 837                (),
 838            );
 839            let exit = Self::notify_internal::<notification::Exit>(&outbound_tx, &());
 840            outbound_tx.close();
 841
 842            let server = self.server.clone();
 843            let name = self.name.clone();
 844            let mut timer = self.executor.timer(SERVER_SHUTDOWN_TIMEOUT).fuse();
 845            Some(
 846                async move {
 847                    log::debug!("language server shutdown started");
 848
 849                    select! {
 850                        request_result = shutdown_request.fuse() => {
 851                            request_result?;
 852                        }
 853
 854                        _ = timer => {
 855                            log::info!("timeout waiting for language server {name} to shutdown");
 856                        },
 857                    }
 858
 859                    response_handlers.lock().take();
 860                    exit?;
 861                    output_done.recv().await;
 862                    server.lock().take().map(|mut child| child.kill());
 863                    log::debug!("language server shutdown finished");
 864
 865                    drop(tasks);
 866                    anyhow::Ok(())
 867                }
 868                .log_err(),
 869            )
 870        } else {
 871            None
 872        }
 873    }
 874
 875    /// Register a handler to handle incoming LSP notifications.
 876    ///
 877    /// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#notificationMessage)
 878    #[must_use]
 879    pub fn on_notification<T, F>(&self, f: F) -> Subscription
 880    where
 881        T: notification::Notification,
 882        F: 'static + Send + FnMut(T::Params, AsyncAppContext),
 883    {
 884        self.on_custom_notification(T::METHOD, f)
 885    }
 886
 887    /// Register a handler to handle incoming LSP requests.
 888    ///
 889    /// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#requestMessage)
 890    #[must_use]
 891    pub fn on_request<T, F, Fut>(&self, f: F) -> Subscription
 892    where
 893        T: request::Request,
 894        T::Params: 'static + Send,
 895        F: 'static + FnMut(T::Params, AsyncAppContext) -> Fut + Send,
 896        Fut: 'static + Future<Output = Result<T::Result>>,
 897    {
 898        self.on_custom_request(T::METHOD, f)
 899    }
 900
 901    /// Registers a handler to inspect all language server process stdio.
 902    #[must_use]
 903    pub fn on_io<F>(&self, f: F) -> Subscription
 904    where
 905        F: 'static + Send + FnMut(IoKind, &str),
 906    {
 907        let id = self.next_id.fetch_add(1, SeqCst);
 908        self.io_handlers.lock().insert(id, Box::new(f));
 909        Subscription::Io {
 910            id,
 911            io_handlers: Some(Arc::downgrade(&self.io_handlers)),
 912        }
 913    }
 914
 915    /// Removes a request handler registers via [`Self::on_request`].
 916    pub fn remove_request_handler<T: request::Request>(&self) {
 917        self.notification_handlers.lock().remove(T::METHOD);
 918    }
 919
 920    /// Removes a notification handler registers via [`Self::on_notification`].
 921    pub fn remove_notification_handler<T: notification::Notification>(&self) {
 922        self.notification_handlers.lock().remove(T::METHOD);
 923    }
 924
 925    /// Checks if a notification handler has been registered via [`Self::on_notification`].
 926    pub fn has_notification_handler<T: notification::Notification>(&self) -> bool {
 927        self.notification_handlers.lock().contains_key(T::METHOD)
 928    }
 929
 930    #[must_use]
 931    fn on_custom_notification<Params, F>(&self, method: &'static str, mut f: F) -> Subscription
 932    where
 933        F: 'static + FnMut(Params, AsyncAppContext) + Send,
 934        Params: DeserializeOwned,
 935    {
 936        let prev_handler = self.notification_handlers.lock().insert(
 937            method,
 938            Box::new(move |_, params, cx| {
 939                if let Some(params) = serde_json::from_value(params).log_err() {
 940                    f(params, cx);
 941                }
 942            }),
 943        );
 944        assert!(
 945            prev_handler.is_none(),
 946            "registered multiple handlers for the same LSP method"
 947        );
 948        Subscription::Notification {
 949            method,
 950            notification_handlers: Some(self.notification_handlers.clone()),
 951        }
 952    }
 953
 954    #[must_use]
 955    fn on_custom_request<Params, Res, Fut, F>(&self, method: &'static str, mut f: F) -> Subscription
 956    where
 957        F: 'static + FnMut(Params, AsyncAppContext) -> Fut + Send,
 958        Fut: 'static + Future<Output = Result<Res>>,
 959        Params: DeserializeOwned + Send + 'static,
 960        Res: Serialize,
 961    {
 962        let outbound_tx = self.outbound_tx.clone();
 963        let prev_handler = self.notification_handlers.lock().insert(
 964            method,
 965            Box::new(move |id, params, cx| {
 966                if let Some(id) = id {
 967                    match serde_json::from_value(params) {
 968                        Ok(params) => {
 969                            let response = f(params, cx.clone());
 970                            cx.foreground_executor()
 971                                .spawn({
 972                                    let outbound_tx = outbound_tx.clone();
 973                                    async move {
 974                                        let response = match response.await {
 975                                            Ok(result) => Response {
 976                                                jsonrpc: JSON_RPC_VERSION,
 977                                                id,
 978                                                value: LspResult::Ok(Some(result)),
 979                                            },
 980                                            Err(error) => Response {
 981                                                jsonrpc: JSON_RPC_VERSION,
 982                                                id,
 983                                                value: LspResult::Error(Some(Error {
 984                                                    message: error.to_string(),
 985                                                })),
 986                                            },
 987                                        };
 988                                        if let Some(response) =
 989                                            serde_json::to_string(&response).log_err()
 990                                        {
 991                                            outbound_tx.try_send(response).ok();
 992                                        }
 993                                    }
 994                                })
 995                                .detach();
 996                        }
 997
 998                        Err(error) => {
 999                            log::error!("error deserializing {} request: {:?}", method, error);
1000                            let response = AnyResponse {
1001                                jsonrpc: JSON_RPC_VERSION,
1002                                id,
1003                                result: None,
1004                                error: Some(Error {
1005                                    message: error.to_string(),
1006                                }),
1007                            };
1008                            if let Some(response) = serde_json::to_string(&response).log_err() {
1009                                outbound_tx.try_send(response).ok();
1010                            }
1011                        }
1012                    }
1013                }
1014            }),
1015        );
1016        assert!(
1017            prev_handler.is_none(),
1018            "registered multiple handlers for the same LSP method"
1019        );
1020        Subscription::Notification {
1021            method,
1022            notification_handlers: Some(self.notification_handlers.clone()),
1023        }
1024    }
1025
1026    /// Get the name of the running language server.
1027    pub fn name(&self) -> LanguageServerName {
1028        self.name.clone()
1029    }
1030
1031    pub fn process_name(&self) -> &str {
1032        &self.process_name
1033    }
1034
1035    /// Get the reported capabilities of the running language server.
1036    pub fn capabilities(&self) -> ServerCapabilities {
1037        self.capabilities.read().clone()
1038    }
1039
1040    /// Get the reported capabilities of the running language server and
1041    /// what we know on the client/adapter-side of its capabilities.
1042    pub fn adapter_server_capabilities(&self) -> AdapterServerCapabilities {
1043        AdapterServerCapabilities {
1044            server_capabilities: self.capabilities(),
1045            code_action_kinds: self.code_action_kinds(),
1046        }
1047    }
1048
1049    pub fn update_capabilities(&self, update: impl FnOnce(&mut ServerCapabilities)) {
1050        update(self.capabilities.write().deref_mut());
1051    }
1052
1053    pub fn configuration(&self) -> &Value {
1054        &self.configuration.settings
1055    }
1056
1057    /// Get the id of the running language server.
1058    pub fn server_id(&self) -> LanguageServerId {
1059        self.server_id
1060    }
1061
1062    /// Language server's binary information.
1063    pub fn binary(&self) -> &LanguageServerBinary {
1064        &self.binary
1065    }
1066    /// Sends a RPC request to the language server.
1067    ///
1068    /// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#requestMessage)
1069    pub fn request<T: request::Request>(
1070        &self,
1071        params: T::Params,
1072    ) -> impl LspRequestFuture<Result<T::Result>>
1073    where
1074        T::Result: 'static + Send,
1075    {
1076        Self::request_internal::<T>(
1077            &self.next_id,
1078            &self.response_handlers,
1079            &self.outbound_tx,
1080            &self.executor,
1081            params,
1082        )
1083    }
1084
1085    fn request_internal<T: request::Request>(
1086        next_id: &AtomicI32,
1087        response_handlers: &Mutex<Option<HashMap<RequestId, ResponseHandler>>>,
1088        outbound_tx: &channel::Sender<String>,
1089        executor: &BackgroundExecutor,
1090        params: T::Params,
1091    ) -> impl LspRequestFuture<Result<T::Result>>
1092    where
1093        T::Result: 'static + Send,
1094    {
1095        let id = next_id.fetch_add(1, SeqCst);
1096        let message = serde_json::to_string(&Request {
1097            jsonrpc: JSON_RPC_VERSION,
1098            id: RequestId::Int(id),
1099            method: T::METHOD,
1100            params,
1101        })
1102        .unwrap();
1103
1104        let (tx, rx) = oneshot::channel();
1105        let handle_response = response_handlers
1106            .lock()
1107            .as_mut()
1108            .ok_or_else(|| anyhow!("server shut down"))
1109            .map(|handlers| {
1110                let executor = executor.clone();
1111                handlers.insert(
1112                    RequestId::Int(id),
1113                    Box::new(move |result| {
1114                        executor
1115                            .spawn(async move {
1116                                let response = match result {
1117                                    Ok(response) => match serde_json::from_str(&response) {
1118                                        Ok(deserialized) => Ok(deserialized),
1119                                        Err(error) => {
1120                                            log::error!("failed to deserialize response from language server: {}. response from language server: {:?}", error, response);
1121                                            Err(error).context("failed to deserialize response")
1122                                        }
1123                                    }
1124                                    Err(error) => Err(anyhow!("{}", error.message)),
1125                                };
1126                                _ = tx.send(response);
1127                            })
1128                            .detach();
1129                    }),
1130                );
1131            });
1132
1133        let send = outbound_tx
1134            .try_send(message)
1135            .context("failed to write to language server's stdin");
1136
1137        let outbound_tx = outbound_tx.downgrade();
1138        let mut timeout = executor.timer(LSP_REQUEST_TIMEOUT).fuse();
1139        let started = Instant::now();
1140        LspRequest::new(id, async move {
1141            handle_response?;
1142            send?;
1143
1144            let cancel_on_drop = util::defer(move || {
1145                if let Some(outbound_tx) = outbound_tx.upgrade() {
1146                    Self::notify_internal::<notification::Cancel>(
1147                        &outbound_tx,
1148                        &CancelParams {
1149                            id: NumberOrString::Number(id),
1150                        },
1151                    )
1152                    .log_err();
1153                }
1154            });
1155
1156            let method = T::METHOD;
1157            select! {
1158                response = rx.fuse() => {
1159                    let elapsed = started.elapsed();
1160                    log::trace!("Took {elapsed:?} to receive response to {method:?} id {id}");
1161                    cancel_on_drop.abort();
1162                    response?
1163                }
1164
1165                _ = timeout => {
1166                    log::error!("Cancelled LSP request task for {method:?} id {id} which took over {LSP_REQUEST_TIMEOUT:?}");
1167                    anyhow::bail!("LSP request timeout");
1168                }
1169            }
1170        })
1171    }
1172
1173    /// Sends a RPC notification to the language server.
1174    ///
1175    /// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#notificationMessage)
1176    pub fn notify<T: notification::Notification>(&self, params: &T::Params) -> Result<()> {
1177        Self::notify_internal::<T>(&self.outbound_tx, params)
1178    }
1179
1180    fn notify_internal<T: notification::Notification>(
1181        outbound_tx: &channel::Sender<String>,
1182        params: &T::Params,
1183    ) -> Result<()> {
1184        let message = serde_json::to_string(&Notification {
1185            jsonrpc: JSON_RPC_VERSION,
1186            method: T::METHOD,
1187            params,
1188        })
1189        .unwrap();
1190        outbound_tx.try_send(message)?;
1191        Ok(())
1192    }
1193
1194    /// Add new workspace folder to the list.
1195    pub fn add_workspace_folder(&self, uri: Url) {
1196        if self
1197            .capabilities()
1198            .workspace
1199            .and_then(|ws| {
1200                ws.workspace_folders.and_then(|folders| {
1201                    folders
1202                        .change_notifications
1203                        .map(|caps| matches!(caps, OneOf::Left(false)))
1204                })
1205            })
1206            .unwrap_or(true)
1207        {
1208            return;
1209        }
1210
1211        let is_new_folder = self.workspace_folders.lock().insert(uri.clone());
1212        if is_new_folder {
1213            let params = DidChangeWorkspaceFoldersParams {
1214                event: WorkspaceFoldersChangeEvent {
1215                    added: vec![WorkspaceFolder {
1216                        uri,
1217                        name: String::default(),
1218                    }],
1219                    removed: vec![],
1220                },
1221            };
1222            self.notify::<DidChangeWorkspaceFolders>(&params).log_err();
1223        }
1224    }
1225    /// Add new workspace folder to the list.
1226    pub fn remove_workspace_folder(&self, uri: Url) {
1227        if self
1228            .capabilities()
1229            .workspace
1230            .and_then(|ws| {
1231                ws.workspace_folders.and_then(|folders| {
1232                    folders
1233                        .change_notifications
1234                        .map(|caps| !matches!(caps, OneOf::Left(false)))
1235                })
1236            })
1237            .unwrap_or(true)
1238        {
1239            return;
1240        }
1241        let was_removed = self.workspace_folders.lock().remove(&uri);
1242        if was_removed {
1243            let params = DidChangeWorkspaceFoldersParams {
1244                event: WorkspaceFoldersChangeEvent {
1245                    added: vec![],
1246                    removed: vec![WorkspaceFolder {
1247                        uri,
1248                        name: String::default(),
1249                    }],
1250                },
1251            };
1252            self.notify::<DidChangeWorkspaceFolders>(&params).log_err();
1253        }
1254    }
1255    pub fn set_workspace_folders(&self, folders: BTreeSet<Url>) {
1256        let mut workspace_folders = self.workspace_folders.lock();
1257        let added: Vec<_> = folders
1258            .iter()
1259            .map(|uri| WorkspaceFolder {
1260                uri: uri.clone(),
1261                name: String::default(),
1262            })
1263            .collect();
1264
1265        let removed: Vec<_> = std::mem::replace(&mut *workspace_folders, folders)
1266            .into_iter()
1267            .map(|uri| WorkspaceFolder {
1268                uri: uri.clone(),
1269                name: String::default(),
1270            })
1271            .collect();
1272        let should_notify = !added.is_empty() || !removed.is_empty();
1273
1274        if should_notify {
1275            let params = DidChangeWorkspaceFoldersParams {
1276                event: WorkspaceFoldersChangeEvent { added, removed },
1277            };
1278            self.notify::<DidChangeWorkspaceFolders>(&params).log_err();
1279        }
1280    }
1281
1282    pub fn workspace_folders(&self) -> impl Deref<Target = BTreeSet<Url>> + '_ {
1283        self.workspace_folders.lock()
1284    }
1285
1286    pub fn register_buffer(
1287        &self,
1288        buffer_id: BufferId,
1289        uri: Url,
1290        language_id: String,
1291        version: i32,
1292        initial_text: String,
1293    ) {
1294        let previous_value = self
1295            .registered_buffers
1296            .lock()
1297            .insert(buffer_id, uri.clone());
1298        if previous_value.is_none() {
1299            self.notify::<notification::DidOpenTextDocument>(&DidOpenTextDocumentParams {
1300                text_document: TextDocumentItem::new(uri, language_id, version, initial_text),
1301            })
1302            .log_err();
1303        } else {
1304            debug_assert_eq!(previous_value, Some(uri));
1305        }
1306    }
1307
1308    pub fn unregister_buffer(&self, buffer_id: BufferId) {
1309        if let Some(path) = self.registered_buffers.lock().remove(&buffer_id) {
1310            self.notify::<notification::DidCloseTextDocument>(&DidCloseTextDocumentParams {
1311                text_document: TextDocumentIdentifier::new(path),
1312            })
1313            .log_err();
1314        }
1315    }
1316}
1317
1318impl Drop for LanguageServer {
1319    fn drop(&mut self) {
1320        if let Some(shutdown) = self.shutdown() {
1321            self.executor.spawn(shutdown).detach();
1322        }
1323    }
1324}
1325
1326impl Subscription {
1327    /// Detaching a subscription handle prevents it from unsubscribing on drop.
1328    pub fn detach(&mut self) {
1329        match self {
1330            Subscription::Notification {
1331                notification_handlers,
1332                ..
1333            } => *notification_handlers = None,
1334            Subscription::Io { io_handlers, .. } => *io_handlers = None,
1335        }
1336    }
1337}
1338
1339impl fmt::Display for LanguageServerId {
1340    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1341        self.0.fmt(f)
1342    }
1343}
1344
1345impl fmt::Debug for LanguageServer {
1346    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1347        f.debug_struct("LanguageServer")
1348            .field("id", &self.server_id.0)
1349            .field("name", &self.name)
1350            .finish_non_exhaustive()
1351    }
1352}
1353
1354impl Drop for Subscription {
1355    fn drop(&mut self) {
1356        match self {
1357            Subscription::Notification {
1358                method,
1359                notification_handlers,
1360            } => {
1361                if let Some(handlers) = notification_handlers {
1362                    handlers.lock().remove(method);
1363                }
1364            }
1365            Subscription::Io { id, io_handlers } => {
1366                if let Some(io_handlers) = io_handlers.as_ref().and_then(|h| h.upgrade()) {
1367                    io_handlers.lock().remove(id);
1368                }
1369            }
1370        }
1371    }
1372}
1373
1374/// Mock language server for use in tests.
1375#[cfg(any(test, feature = "test-support"))]
1376#[derive(Clone)]
1377pub struct FakeLanguageServer {
1378    pub binary: LanguageServerBinary,
1379    pub server: Arc<LanguageServer>,
1380    notifications_rx: channel::Receiver<(String, String)>,
1381}
1382
1383#[cfg(any(test, feature = "test-support"))]
1384impl FakeLanguageServer {
1385    /// Construct a fake language server.
1386    pub fn new(
1387        server_id: LanguageServerId,
1388        binary: LanguageServerBinary,
1389        name: String,
1390        capabilities: ServerCapabilities,
1391        cx: AsyncAppContext,
1392    ) -> (LanguageServer, FakeLanguageServer) {
1393        let (stdin_writer, stdin_reader) = async_pipe::pipe();
1394        let (stdout_writer, stdout_reader) = async_pipe::pipe();
1395        let (notifications_tx, notifications_rx) = channel::unbounded();
1396
1397        let server_name = LanguageServerName(name.clone().into());
1398        let process_name = Arc::from(name.as_str());
1399        let mut server = LanguageServer::new_internal(
1400            server_id,
1401            server_name.clone(),
1402            stdin_writer,
1403            stdout_reader,
1404            None::<async_pipe::PipeReader>,
1405            Arc::new(Mutex::new(None)),
1406            None,
1407            None,
1408            binary.clone(),
1409            cx.clone(),
1410            |_| {},
1411        );
1412        server.process_name = process_name;
1413        let fake = FakeLanguageServer {
1414            binary: binary.clone(),
1415            server: Arc::new({
1416                let mut server = LanguageServer::new_internal(
1417                    server_id,
1418                    server_name,
1419                    stdout_writer,
1420                    stdin_reader,
1421                    None::<async_pipe::PipeReader>,
1422                    Arc::new(Mutex::new(None)),
1423                    None,
1424                    None,
1425                    binary,
1426                    cx.clone(),
1427                    move |msg| {
1428                        notifications_tx
1429                            .try_send((
1430                                msg.method.to_string(),
1431                                msg.params.unwrap_or(Value::Null).to_string(),
1432                            ))
1433                            .ok();
1434                    },
1435                );
1436                server.process_name = name.as_str().into();
1437                server
1438            }),
1439            notifications_rx,
1440        };
1441        fake.handle_request::<request::Initialize, _, _>({
1442            let capabilities = capabilities;
1443            move |_, _| {
1444                let capabilities = capabilities.clone();
1445                let name = name.clone();
1446                async move {
1447                    Ok(InitializeResult {
1448                        capabilities,
1449                        server_info: Some(ServerInfo {
1450                            name,
1451                            ..Default::default()
1452                        }),
1453                    })
1454                }
1455            }
1456        });
1457
1458        (server, fake)
1459    }
1460}
1461
1462#[cfg(any(test, feature = "test-support"))]
1463impl LanguageServer {
1464    pub fn full_capabilities() -> ServerCapabilities {
1465        ServerCapabilities {
1466            document_highlight_provider: Some(OneOf::Left(true)),
1467            code_action_provider: Some(CodeActionProviderCapability::Simple(true)),
1468            document_formatting_provider: Some(OneOf::Left(true)),
1469            document_range_formatting_provider: Some(OneOf::Left(true)),
1470            definition_provider: Some(OneOf::Left(true)),
1471            implementation_provider: Some(ImplementationProviderCapability::Simple(true)),
1472            type_definition_provider: Some(TypeDefinitionProviderCapability::Simple(true)),
1473            ..Default::default()
1474        }
1475    }
1476}
1477
1478#[cfg(any(test, feature = "test-support"))]
1479impl FakeLanguageServer {
1480    /// See [`LanguageServer::notify`].
1481    pub fn notify<T: notification::Notification>(&self, params: &T::Params) {
1482        self.server.notify::<T>(params).ok();
1483    }
1484
1485    /// See [`LanguageServer::request`].
1486    pub async fn request<T>(&self, params: T::Params) -> Result<T::Result>
1487    where
1488        T: request::Request,
1489        T::Result: 'static + Send,
1490    {
1491        self.server.executor.start_waiting();
1492        self.server.request::<T>(params).await
1493    }
1494
1495    /// Attempts [`Self::try_receive_notification`], unwrapping if it has not received the specified type yet.
1496    pub async fn receive_notification<T: notification::Notification>(&mut self) -> T::Params {
1497        self.server.executor.start_waiting();
1498        self.try_receive_notification::<T>().await.unwrap()
1499    }
1500
1501    /// Consumes the notification channel until it finds a notification for the specified type.
1502    pub async fn try_receive_notification<T: notification::Notification>(
1503        &mut self,
1504    ) -> Option<T::Params> {
1505        loop {
1506            let (method, params) = self.notifications_rx.recv().await.ok()?;
1507            if method == T::METHOD {
1508                return Some(serde_json::from_str::<T::Params>(&params).unwrap());
1509            } else {
1510                log::info!("skipping message in fake language server {:?}", params);
1511            }
1512        }
1513    }
1514
1515    /// Registers a handler for a specific kind of request. Removes any existing handler for specified request type.
1516    pub fn handle_request<T, F, Fut>(
1517        &self,
1518        mut handler: F,
1519    ) -> futures::channel::mpsc::UnboundedReceiver<()>
1520    where
1521        T: 'static + request::Request,
1522        T::Params: 'static + Send,
1523        F: 'static + Send + FnMut(T::Params, gpui::AsyncAppContext) -> Fut,
1524        Fut: 'static + Send + Future<Output = Result<T::Result>>,
1525    {
1526        let (responded_tx, responded_rx) = futures::channel::mpsc::unbounded();
1527        self.server.remove_request_handler::<T>();
1528        self.server
1529            .on_request::<T, _, _>(move |params, cx| {
1530                let result = handler(params, cx.clone());
1531                let responded_tx = responded_tx.clone();
1532                let executor = cx.background_executor().clone();
1533                async move {
1534                    executor.simulate_random_delay().await;
1535                    let result = result.await;
1536                    responded_tx.unbounded_send(()).ok();
1537                    result
1538                }
1539            })
1540            .detach();
1541        responded_rx
1542    }
1543
1544    /// Registers a handler for a specific kind of notification. Removes any existing handler for specified notification type.
1545    pub fn handle_notification<T, F>(
1546        &self,
1547        mut handler: F,
1548    ) -> futures::channel::mpsc::UnboundedReceiver<()>
1549    where
1550        T: 'static + notification::Notification,
1551        T::Params: 'static + Send,
1552        F: 'static + Send + FnMut(T::Params, gpui::AsyncAppContext),
1553    {
1554        let (handled_tx, handled_rx) = futures::channel::mpsc::unbounded();
1555        self.server.remove_notification_handler::<T>();
1556        self.server
1557            .on_notification::<T, _>(move |params, cx| {
1558                handler(params, cx.clone());
1559                handled_tx.unbounded_send(()).ok();
1560            })
1561            .detach();
1562        handled_rx
1563    }
1564
1565    /// Removes any existing handler for specified notification type.
1566    pub fn remove_request_handler<T>(&mut self)
1567    where
1568        T: 'static + request::Request,
1569    {
1570        self.server.remove_request_handler::<T>();
1571    }
1572
1573    /// Simulate that the server has started work and notifies about its progress with the specified token.
1574    pub async fn start_progress(&self, token: impl Into<String>) {
1575        self.start_progress_with(token, Default::default()).await
1576    }
1577
1578    pub async fn start_progress_with(
1579        &self,
1580        token: impl Into<String>,
1581        progress: WorkDoneProgressBegin,
1582    ) {
1583        let token = token.into();
1584        self.request::<request::WorkDoneProgressCreate>(WorkDoneProgressCreateParams {
1585            token: NumberOrString::String(token.clone()),
1586        })
1587        .await
1588        .unwrap();
1589        self.notify::<notification::Progress>(&ProgressParams {
1590            token: NumberOrString::String(token),
1591            value: ProgressParamsValue::WorkDone(WorkDoneProgress::Begin(progress)),
1592        });
1593    }
1594
1595    /// Simulate that the server has completed work and notifies about that with the specified token.
1596    pub fn end_progress(&self, token: impl Into<String>) {
1597        self.notify::<notification::Progress>(&ProgressParams {
1598            token: NumberOrString::String(token.into()),
1599            value: ProgressParamsValue::WorkDone(WorkDoneProgress::End(Default::default())),
1600        });
1601    }
1602}
1603
1604#[cfg(test)]
1605mod tests {
1606    use super::*;
1607    use gpui::{SemanticVersion, TestAppContext};
1608    use std::str::FromStr;
1609
1610    #[ctor::ctor]
1611    fn init_logger() {
1612        if std::env::var("RUST_LOG").is_ok() {
1613            env_logger::init();
1614        }
1615    }
1616
1617    #[gpui::test]
1618    async fn test_fake(cx: &mut TestAppContext) {
1619        cx.update(|cx| {
1620            release_channel::init(SemanticVersion::default(), cx);
1621        });
1622        let (server, mut fake) = FakeLanguageServer::new(
1623            LanguageServerId(0),
1624            LanguageServerBinary {
1625                path: "path/to/language-server".into(),
1626                arguments: vec![],
1627                env: None,
1628            },
1629            "the-lsp".to_string(),
1630            Default::default(),
1631            cx.to_async(),
1632        );
1633
1634        let (message_tx, message_rx) = channel::unbounded();
1635        let (diagnostics_tx, diagnostics_rx) = channel::unbounded();
1636        server
1637            .on_notification::<notification::ShowMessage, _>(move |params, _| {
1638                message_tx.try_send(params).unwrap()
1639            })
1640            .detach();
1641        server
1642            .on_notification::<notification::PublishDiagnostics, _>(move |params, _| {
1643                diagnostics_tx.try_send(params).unwrap()
1644            })
1645            .detach();
1646
1647        let server = cx
1648            .update(|cx| {
1649                let params = server.default_initialize_params(cx);
1650                let configuration = DidChangeConfigurationParams {
1651                    settings: Default::default(),
1652                };
1653                server.initialize(params, configuration.into(), cx)
1654            })
1655            .await
1656            .unwrap();
1657        server
1658            .notify::<notification::DidOpenTextDocument>(&DidOpenTextDocumentParams {
1659                text_document: TextDocumentItem::new(
1660                    Url::from_str("file://a/b").unwrap(),
1661                    "rust".to_string(),
1662                    0,
1663                    "".to_string(),
1664                ),
1665            })
1666            .unwrap();
1667        assert_eq!(
1668            fake.receive_notification::<notification::DidOpenTextDocument>()
1669                .await
1670                .text_document
1671                .uri
1672                .as_str(),
1673            "file://a/b"
1674        );
1675
1676        fake.notify::<notification::ShowMessage>(&ShowMessageParams {
1677            typ: MessageType::ERROR,
1678            message: "ok".to_string(),
1679        });
1680        fake.notify::<notification::PublishDiagnostics>(&PublishDiagnosticsParams {
1681            uri: Url::from_str("file://b/c").unwrap(),
1682            version: Some(5),
1683            diagnostics: vec![],
1684        });
1685        assert_eq!(message_rx.recv().await.unwrap().message, "ok");
1686        assert_eq!(
1687            diagnostics_rx.recv().await.unwrap().uri.as_str(),
1688            "file://b/c"
1689        );
1690
1691        fake.handle_request::<request::Shutdown, _, _>(|_, _| async move { Ok(()) });
1692
1693        drop(server);
1694        fake.receive_notification::<notification::Exit>().await;
1695    }
1696
1697    #[gpui::test]
1698    fn test_deserialize_string_digit_id() {
1699        let json = r#"{"jsonrpc":"2.0","id":"2","method":"workspace/configuration","params":{"items":[{"scopeUri":"file:///Users/mph/Devel/personal/hello-scala/","section":"metals"}]}}"#;
1700        let notification = serde_json::from_str::<AnyNotification>(json)
1701            .expect("message with string id should be parsed");
1702        let expected_id = RequestId::Str("2".to_string());
1703        assert_eq!(notification.id, Some(expected_id));
1704    }
1705
1706    #[gpui::test]
1707    fn test_deserialize_string_id() {
1708        let json = r#"{"jsonrpc":"2.0","id":"anythingAtAll","method":"workspace/configuration","params":{"items":[{"scopeUri":"file:///Users/mph/Devel/personal/hello-scala/","section":"metals"}]}}"#;
1709        let notification = serde_json::from_str::<AnyNotification>(json)
1710            .expect("message with string id should be parsed");
1711        let expected_id = RequestId::Str("anythingAtAll".to_string());
1712        assert_eq!(notification.id, Some(expected_id));
1713    }
1714
1715    #[gpui::test]
1716    fn test_deserialize_int_id() {
1717        let json = r#"{"jsonrpc":"2.0","id":2,"method":"workspace/configuration","params":{"items":[{"scopeUri":"file:///Users/mph/Devel/personal/hello-scala/","section":"metals"}]}}"#;
1718        let notification = serde_json::from_str::<AnyNotification>(json)
1719            .expect("message with string id should be parsed");
1720        let expected_id = RequestId::Int(2);
1721        assert_eq!(notification.id, Some(expected_id));
1722    }
1723
1724    #[test]
1725    fn test_serialize_has_no_nulls() {
1726        // Ensure we're not setting both result and error variants. (ticket #10595)
1727        let no_tag = Response::<u32> {
1728            jsonrpc: "",
1729            id: RequestId::Int(0),
1730            value: LspResult::Ok(None),
1731        };
1732        assert_eq!(
1733            serde_json::to_string(&no_tag).unwrap(),
1734            "{\"jsonrpc\":\"\",\"id\":0,\"result\":null}"
1735        );
1736        let no_tag = Response::<u32> {
1737            jsonrpc: "",
1738            id: RequestId::Int(0),
1739            value: LspResult::Error(None),
1740        };
1741        assert_eq!(
1742            serde_json::to_string(&no_tag).unwrap(),
1743            "{\"jsonrpc\":\"\",\"id\":0,\"error\":null}"
1744        );
1745    }
1746}