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