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