mod input_handler; pub use lsp_types::request::*;pub use lsp_types::*; use anyhow::{anyhow, Context, Result};use collections::HashMap;use futures::{channel::oneshot, io::BufWriter, select, AsyncRead, AsyncWrite, Future, FutureExt};use gpui::{AppContext, AsyncAppContext, BackgroundExecutor, SharedString, Task};use parking_lot::{Mutex, RwLock};use postage::{barrier, prelude::Stream};use schemars::{ gen::SchemaGenerator, schema::{InstanceType, Schema, SchemaObject}, JsonSchema,};use serde::{de::DeserializeOwned, Deserialize, Serialize};use serde_json::{json, value::RawValue, Value};use smol::{ channel, io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, process::Child,}; use std::{ ffi::{OsStr, OsString}, fmt, io::Write, ops::DerefMut, path::PathBuf, pin::Pin, sync::{ atomic::{AtomicI32, Ordering::SeqCst}, Arc, Weak, }, task::Poll, time::{Duration, Instant},};use std::{path::Path, process::Stdio};use util::{ResultExt, TryFutureExt}; const JSON_RPC_VERSION: &str = "2.0";const CONTENT_LEN_HEADER: &str = "Content-Length: "; const LSP_REQUEST_TIMEOUT: Duration = Duration::from_secs(60 * 2);const SERVER_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5); type NotificationHandler = Box<dyn Send + FnMut(Option<RequestId>, Value, AsyncAppContext)>;type ResponseHandler = Box<dyn Send + FnOnce(Result<String, Error>)>;type IoHandler = Box<dyn Send + FnMut(IoKind, &str)>; /// Kind of language server stdio given to an IO handler.#[derive(Debug, Clone, Copy)]pub enum IoKind { StdOut, StdIn, StdErr,} /// Represents a launchable language server. This can either be a standalone binary or the path/// to a runtime with arguments to instruct it to launch the actual language server file.#[derive(Debug, Clone, Deserialize)]pub struct LanguageServerBinary { pub path: PathBuf, pub arguments: Vec<OsString>, pub env: Option<HashMap<String, String>>,} /// Configures the search (and installation) of language servers.#[derive(Debug, Clone, Deserialize)]pub struct LanguageServerBinaryOptions { /// Whether the adapter should look at the users system pub allow_path_lookup: bool, /// Whether the adapter should download its own version pub allow_binary_download: bool,} /// A running language server process.pub struct LanguageServer { server_id: LanguageServerId, next_id: AtomicI32, outbound_tx: channel::Sender<String>, name: LanguageServerName, process_name: Arc<str>, capabilities: RwLock<ServerCapabilities>, code_action_kinds: Option<Vec<CodeActionKind>>, notification_handlers: Arc<Mutex<HashMap<&'static str, NotificationHandler>>>, response_handlers: Arc<Mutex<Option<HashMap<RequestId, ResponseHandler>>>>, io_handlers: Arc<Mutex<HashMap<i32, IoHandler>>>, executor: BackgroundExecutor, #[allow(clippy::type_complexity)] io_tasks: Mutex<Option<(Task<Option<()>>, Task<Option<()>>)>>, output_done_rx: Mutex<Option<barrier::Receiver>>, root_path: PathBuf, working_dir: PathBuf, server: Arc<Mutex<Option<Child>>>,} /// Identifies a running language server.#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]#[repr(transparent)]pub struct LanguageServerId(pub usize); impl LanguageServerId { pub fn from_proto(id: u64) -> Self { Self(id as usize) } pub fn to_proto(self) -> u64 { self.0 as u64 }} /// A name of a language server.#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Deserialize, Serialize)]pub struct LanguageServerName(pub SharedString); impl std::fmt::Display for LanguageServerName { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { std::fmt::Display::fmt(&self.0, f) }} impl AsRef<str> for LanguageServerName { fn as_ref(&self) -> &str { self.0.as_ref() }} impl AsRef<OsStr> for LanguageServerName { fn as_ref(&self) -> &OsStr { self.0.as_ref().as_ref() }} impl JsonSchema for LanguageServerName { fn schema_name() -> String { "LanguageServerName".into() } fn json_schema(_: &mut SchemaGenerator) -> Schema { SchemaObject { instance_type: Some(InstanceType::String.into()), ..Default::default() } .into() }} impl LanguageServerName { pub const fn new_static(s: &'static str) -> Self { Self(SharedString::new_static(s)) } pub fn from_proto(s: String) -> Self { Self(s.into()) }} impl<'a> From<&'a str> for LanguageServerName { fn from(str: &'a str) -> LanguageServerName { LanguageServerName(str.to_string().into()) }} /// Handle to a language server RPC activity subscription.pub enum Subscription { Notification { method: &'static str, notification_handlers: Option<Arc<Mutex<HashMap<&'static str, NotificationHandler>>>>, }, Io { id: i32, io_handlers: Option<Weak<Mutex<HashMap<i32, IoHandler>>>>, },} /// Language server protocol RPC request message ID.////// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#requestMessage)#[derive(Debug, Clone, Eq, PartialEq, Hash, Serialize, Deserialize)]#[serde(untagged)]pub enum RequestId { Int(i32), Str(String),} /// Language server protocol RPC request message.////// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#requestMessage)#[derive(Serialize, Deserialize)]pub struct Request<'a, T> { jsonrpc: &'static str, id: RequestId, method: &'a str, params: T,} /// Language server protocol RPC request response message before it is deserialized into a concrete type.#[derive(Serialize, Deserialize)]struct AnyResponse<'a> { jsonrpc: &'a str, id: RequestId, #[serde(default)] error: Option<Error>, #[serde(borrow)] result: Option<&'a RawValue>,} /// Language server protocol RPC request response message.////// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#responseMessage)#[derive(Serialize)]struct Response<T> { jsonrpc: &'static str, id: RequestId, #[serde(flatten)] value: LspResult<T>,} #[derive(Serialize)]#[serde(rename_all = "snake_case")]enum LspResult<T> { #[serde(rename = "result")] Ok(Option<T>), Error(Option<Error>),} /// Language server protocol RPC notification message.////// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#notificationMessage)#[derive(Serialize, Deserialize)]struct Notification<'a, T> { jsonrpc: &'static str, #[serde(borrow)] method: &'a str, params: T,} /// Language server RPC notification message before it is deserialized into a concrete type.#[derive(Debug, Clone, Deserialize)]struct AnyNotification { #[serde(default)] id: Option<RequestId>, method: String, #[serde(default)] params: Option<Value>,} #[derive(Debug, Serialize, Deserialize)]struct Error { message: String,} pub trait LspRequestFuture<O>: Future<Output = O> { fn id(&self) -> i32;} struct LspRequest<F> { id: i32, request: F,} impl<F> LspRequest<F> { pub fn new(id: i32, request: F) -> Self { Self { id, request } }} impl<F: Future> Future for LspRequest<F> { type Output = F::Output; fn poll(self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> { // SAFETY: This is standard pin projection, we're pinned so our fields must be pinned. let inner = unsafe { Pin::new_unchecked(&mut self.get_unchecked_mut().request) }; inner.poll(cx) }} impl<F: Future> LspRequestFuture<F::Output> for LspRequest<F> { fn id(&self) -> i32 { self.id }} /// Combined capabilities of the server and the adapter.pub struct AdapterServerCapabilities { // Reported capabilities by the server pub server_capabilities: ServerCapabilities, // List of code actions supported by the LspAdapter matching the server pub code_action_kinds: Option<Vec<CodeActionKind>>,} /// Experimental: Informs the end user about the state of the server////// [Rust Analyzer Specification](https://github.com/rust-lang/rust-analyzer/blob/master/docs/dev/lsp-extensions.md#server-status)#[derive(Debug)]pub enum ServerStatus {} /// Other(String) variant to handle unknown values due to this still being experimental#[derive(Debug, PartialEq, Deserialize, Serialize, Clone)]#[serde(rename_all = "camelCase")]pub enum ServerHealthStatus { Ok, Warning, Error, Other(String),} #[derive(Debug, PartialEq, Deserialize, Serialize, Clone)]#[serde(rename_all = "camelCase")]pub struct ServerStatusParams { pub health: ServerHealthStatus, pub message: Option<String>,} impl lsp_types::notification::Notification for ServerStatus { type Params = ServerStatusParams; const METHOD: &'static str = "experimental/serverStatus";} impl LanguageServer { /// Starts a language server process. pub fn new( stderr_capture: Arc<Mutex<Option<String>>>, server_id: LanguageServerId, server_name: LanguageServerName, binary: LanguageServerBinary, root_path: &Path, code_action_kinds: Option<Vec<CodeActionKind>>, cx: AsyncAppContext, ) -> Result<Self> { let working_dir = if root_path.is_dir() { root_path } else { root_path.parent().unwrap_or_else(|| Path::new("/")) }; log::info!( "starting language server process. binary path: {:?}, working directory: {:?}, args: {:?}", binary.path, working_dir, &binary.arguments ); let mut server = util::command::new_smol_command(&binary.path) .current_dir(working_dir) .args(&binary.arguments) .envs(binary.env.unwrap_or_default()) .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::piped()) .kill_on_drop(true) .spawn() .with_context(|| { format!( "failed to spawn command. path: {:?}, working directory: {:?}, args: {:?}", binary.path, working_dir, &binary.arguments ) })?; let stdin = server.stdin.take().unwrap(); let stdout = server.stdout.take().unwrap(); let stderr = server.stderr.take().unwrap(); let mut server = Self::new_internal( server_id, server_name, stdin, stdout, Some(stderr), stderr_capture, Some(server), root_path, working_dir, code_action_kinds, cx, move |notification| { log::info!( "Language server with id {} sent unhandled notification {}:\n{}", server_id, notification.method, serde_json::to_string_pretty(¬ification.params).unwrap(), ); }, ); if let Some(name) = binary.path.file_name() { server.process_name = name.to_string_lossy().into(); } Ok(server) } #[allow(clippy::too_many_arguments)] fn new_internal<Stdin, Stdout, Stderr, F>( server_id: LanguageServerId, server_name: LanguageServerName, stdin: Stdin, stdout: Stdout, stderr: Option<Stderr>, stderr_capture: Arc<Mutex<Option<String>>>, server: Option<Child>, root_path: &Path, working_dir: &Path, code_action_kinds: Option<Vec<CodeActionKind>>, cx: AsyncAppContext, on_unhandled_notification: F, ) -> Self where Stdin: AsyncWrite + Unpin + Send + 'static, Stdout: AsyncRead + Unpin + Send + 'static, Stderr: AsyncRead + Unpin + Send + 'static, F: FnMut(AnyNotification) + 'static + Send + Sync + Clone, { let (outbound_tx, outbound_rx) = channel::unbounded::<String>(); let (output_done_tx, output_done_rx) = barrier::channel(); let notification_handlers = Arc::new(Mutex::new(HashMap::<_, NotificationHandler>::default())); let response_handlers = Arc::new(Mutex::new(Some(HashMap::<_, ResponseHandler>::default()))); let io_handlers = Arc::new(Mutex::new(HashMap::default())); let stdout_input_task = cx.spawn({ let on_unhandled_notification = on_unhandled_notification.clone(); let notification_handlers = notification_handlers.clone(); let response_handlers = response_handlers.clone(); let io_handlers = io_handlers.clone(); move |cx| { Self::handle_input( stdout, on_unhandled_notification, notification_handlers, response_handlers, io_handlers, cx, ) .log_err() } }); let stderr_input_task = stderr .map(|stderr| { let io_handlers = io_handlers.clone(); let stderr_captures = stderr_capture.clone(); cx.spawn(|_| Self::handle_stderr(stderr, io_handlers, stderr_captures).log_err()) }) .unwrap_or_else(|| Task::Ready(Some(None))); let input_task = cx.spawn(|_| async move { let (stdout, stderr) = futures::join!(stdout_input_task, stderr_input_task); stdout.or(stderr) }); let output_task = cx.background_executor().spawn({ Self::handle_output( stdin, outbound_rx, output_done_tx, response_handlers.clone(), io_handlers.clone(), ) .log_err() }); Self { server_id, notification_handlers, response_handlers, io_handlers, name: server_name, process_name: Arc::default(), capabilities: Default::default(), code_action_kinds, next_id: Default::default(), outbound_tx, executor: cx.background_executor().clone(), io_tasks: Mutex::new(Some((input_task, output_task))), output_done_rx: Mutex::new(Some(output_done_rx)), root_path: root_path.to_path_buf(), working_dir: working_dir.to_path_buf(), server: Arc::new(Mutex::new(server)), } } /// List of code action kinds this language server reports being able to emit. pub fn code_action_kinds(&self) -> Option<Vec<CodeActionKind>> { self.code_action_kinds.clone() } async fn handle_input<Stdout, F>( stdout: Stdout, mut on_unhandled_notification: F, notification_handlers: Arc<Mutex<HashMap<&'static str, NotificationHandler>>>, response_handlers: Arc<Mutex<Option<HashMap<RequestId, ResponseHandler>>>>, io_handlers: Arc<Mutex<HashMap<i32, IoHandler>>>, cx: AsyncAppContext, ) -> anyhow::Result<()> where Stdout: AsyncRead + Unpin + Send + 'static, F: FnMut(AnyNotification) + 'static + Send, { use smol::stream::StreamExt; let stdout = BufReader::new(stdout);Lines 1–500 of 1621. Load more
mod input_handler; pub use lsp_types::request::*; pub use lsp_types::*; use anyhow::{anyhow, Context, Result}; use collections::HashMap; use futures::{channel::oneshot, io::BufWriter, select, AsyncRead, AsyncWrite, Future, FutureExt}; use gpui::{AppContext, AsyncAppContext, BackgroundExecutor, SharedString, Task}; use parking_lot::{Mutex, RwLock}; use postage::{barrier, prelude::Stream}; use schemars::{ gen::SchemaGenerator, schema::{InstanceType, Schema, SchemaObject}, JsonSchema, }; use serde::{de::DeserializeOwned, Deserialize, Serialize}; use serde_json::{json, value::RawValue, Value}; use smol::{ channel, io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, process::Child, }; use std::{ ffi::{OsStr, OsString}, fmt, io::Write, ops::DerefMut, path::PathBuf, pin::Pin, sync::{ atomic::{AtomicI32, Ordering::SeqCst}, Arc, Weak, }, task::Poll, time::{Duration, Instant}, }; use std::{path::Path, process::Stdio}; use util::{ResultExt, TryFutureExt}; const JSON_RPC_VERSION: &str = "2.0"; const CONTENT_LEN_HEADER: &str = "Content-Length: "; const LSP_REQUEST_TIMEOUT: Duration = Duration::from_secs(60 * 2); const SERVER_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5); type NotificationHandler = Box<dyn Send + FnMut(Option<RequestId>, Value, AsyncAppContext)>; type ResponseHandler = Box<dyn Send + FnOnce(Result<String, Error>)>; type IoHandler = Box<dyn Send + FnMut(IoKind, &str)>; /// Kind of language server stdio given to an IO handler. #[derive(Debug, Clone, Copy)] pub enum IoKind { StdOut, StdIn, StdErr, } /// Represents a launchable language server. This can either be a standalone binary or the path /// to a runtime with arguments to instruct it to launch the actual language server file. #[derive(Debug, Clone, Deserialize)] pub struct LanguageServerBinary { pub path: PathBuf, pub arguments: Vec<OsString>, pub env: Option<HashMap<String, String>>, } /// Configures the search (and installation) of language servers. #[derive(Debug, Clone, Deserialize)] pub struct LanguageServerBinaryOptions { /// Whether the adapter should look at the users system pub allow_path_lookup: bool, /// Whether the adapter should download its own version pub allow_binary_download: bool, } /// A running language server process. pub struct LanguageServer { server_id: LanguageServerId, next_id: AtomicI32, outbound_tx: channel::Sender<String>, name: LanguageServerName, process_name: Arc<str>, capabilities: RwLock<ServerCapabilities>, code_action_kinds: Option<Vec<CodeActionKind>>, notification_handlers: Arc<Mutex<HashMap<&'static str, NotificationHandler>>>, response_handlers: Arc<Mutex<Option<HashMap<RequestId, ResponseHandler>>>>, io_handlers: Arc<Mutex<HashMap<i32, IoHandler>>>, executor: BackgroundExecutor, #[allow(clippy::type_complexity)] io_tasks: Mutex<Option<(Task<Option<()>>, Task<Option<()>>)>>, output_done_rx: Mutex<Option<barrier::Receiver>>, root_path: PathBuf, working_dir: PathBuf, server: Arc<Mutex<Option<Child>>>, } /// Identifies a running language server. #[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] #[repr(transparent)] pub struct LanguageServerId(pub usize); impl LanguageServerId { pub fn from_proto(id: u64) -> Self { Self(id as usize) } pub fn to_proto(self) -> u64 { self.0 as u64 } } /// A name of a language server. #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Deserialize, Serialize)] pub struct LanguageServerName(pub SharedString); impl std::fmt::Display for LanguageServerName { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { std::fmt::Display::fmt(&self.0, f) } } impl AsRef<str> for LanguageServerName { fn as_ref(&self) -> &str { self.0.as_ref() } } impl AsRef<OsStr> for LanguageServerName { fn as_ref(&self) -> &OsStr { self.0.as_ref().as_ref() } } impl JsonSchema for LanguageServerName { fn schema_name() -> String { "LanguageServerName".into() } fn json_schema(_: &mut SchemaGenerator) -> Schema { SchemaObject { instance_type: Some(InstanceType::String.into()), ..Default::default() } .into() } } impl LanguageServerName { pub const fn new_static(s: &'static str) -> Self { Self(SharedString::new_static(s)) } pub fn from_proto(s: String) -> Self { Self(s.into()) } } impl<'a> From<&'a str> for LanguageServerName { fn from(str: &'a str) -> LanguageServerName { LanguageServerName(str.to_string().into()) } } /// Handle to a language server RPC activity subscription. pub enum Subscription { Notification { method: &'static str, notification_handlers: Option<Arc<Mutex<HashMap<&'static str, NotificationHandler>>>>, }, Io { id: i32, io_handlers: Option<Weak<Mutex<HashMap<i32, IoHandler>>>>, }, } /// Language server protocol RPC request message ID. /// /// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#requestMessage) #[derive(Debug, Clone, Eq, PartialEq, Hash, Serialize, Deserialize)] #[serde(untagged)] pub enum RequestId { Int(i32), Str(String), } /// Language server protocol RPC request message. /// /// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#requestMessage) #[derive(Serialize, Deserialize)] pub struct Request<'a, T> { jsonrpc: &'static str, id: RequestId, method: &'a str, params: T, } /// Language server protocol RPC request response message before it is deserialized into a concrete type. #[derive(Serialize, Deserialize)] struct AnyResponse<'a> { jsonrpc: &'a str, id: RequestId, #[serde(default)] error: Option<Error>, #[serde(borrow)] result: Option<&'a RawValue>, } /// Language server protocol RPC request response message. /// /// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#responseMessage) #[derive(Serialize)] struct Response<T> { jsonrpc: &'static str, id: RequestId, #[serde(flatten)] value: LspResult<T>, } #[derive(Serialize)] #[serde(rename_all = "snake_case")] enum LspResult<T> { #[serde(rename = "result")] Ok(Option<T>), Error(Option<Error>), } /// Language server protocol RPC notification message. /// /// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#notificationMessage) #[derive(Serialize, Deserialize)] struct Notification<'a, T> { jsonrpc: &'static str, #[serde(borrow)] method: &'a str, params: T, } /// Language server RPC notification message before it is deserialized into a concrete type. #[derive(Debug, Clone, Deserialize)] struct AnyNotification { #[serde(default)] id: Option<RequestId>, method: String, #[serde(default)] params: Option<Value>, } #[derive(Debug, Serialize, Deserialize)] struct Error { message: String, } pub trait LspRequestFuture<O>: Future<Output = O> { fn id(&self) -> i32; } struct LspRequest<F> { id: i32, request: F, } impl<F> LspRequest<F> { pub fn new(id: i32, request: F) -> Self { Self { id, request } } } impl<F: Future> Future for LspRequest<F> { type Output = F::Output; fn poll(self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> { // SAFETY: This is standard pin projection, we're pinned so our fields must be pinned. let inner = unsafe { Pin::new_unchecked(&mut self.get_unchecked_mut().request) }; inner.poll(cx) } } impl<F: Future> LspRequestFuture<F::Output> for LspRequest<F> { fn id(&self) -> i32 { self.id } } /// Combined capabilities of the server and the adapter. pub struct AdapterServerCapabilities { // Reported capabilities by the server pub server_capabilities: ServerCapabilities, // List of code actions supported by the LspAdapter matching the server pub code_action_kinds: Option<Vec<CodeActionKind>>, } /// Experimental: Informs the end user about the state of the server /// /// [Rust Analyzer Specification](https://github.com/rust-lang/rust-analyzer/blob/master/docs/dev/lsp-extensions.md#server-status) #[derive(Debug)] pub enum ServerStatus {} /// Other(String) variant to handle unknown values due to this still being experimental #[derive(Debug, PartialEq, Deserialize, Serialize, Clone)] #[serde(rename_all = "camelCase")] pub enum ServerHealthStatus { Ok, Warning, Error, Other(String), } #[derive(Debug, PartialEq, Deserialize, Serialize, Clone)] #[serde(rename_all = "camelCase")] pub struct ServerStatusParams { pub health: ServerHealthStatus, pub message: Option<String>, } impl lsp_types::notification::Notification for ServerStatus { type Params = ServerStatusParams; const METHOD: &'static str = "experimental/serverStatus"; } impl LanguageServer { /// Starts a language server process. pub fn new( stderr_capture: Arc<Mutex<Option<String>>>, server_id: LanguageServerId, server_name: LanguageServerName, binary: LanguageServerBinary, root_path: &Path, code_action_kinds: Option<Vec<CodeActionKind>>, cx: AsyncAppContext, ) -> Result<Self> { let working_dir = if root_path.is_dir() { root_path } else { root_path.parent().unwrap_or_else(|| Path::new("/")) }; log::info!( "starting language server process. binary path: {:?}, working directory: {:?}, args: {:?}", binary.path, working_dir, &binary.arguments ); let mut server = util::command::new_smol_command(&binary.path) .current_dir(working_dir) .args(&binary.arguments) .envs(binary.env.unwrap_or_default()) .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::piped()) .kill_on_drop(true) .spawn() .with_context(|| { format!( "failed to spawn command. path: {:?}, working directory: {:?}, args: {:?}", binary.path, working_dir, &binary.arguments ) })?; let stdin = server.stdin.take().unwrap(); let stdout = server.stdout.take().unwrap(); let stderr = server.stderr.take().unwrap(); let mut server = Self::new_internal( server_id, server_name, stdin, stdout, Some(stderr), stderr_capture, Some(server), root_path, working_dir, code_action_kinds, cx, move |notification| { log::info!( "Language server with id {} sent unhandled notification {}:\n{}", server_id, notification.method, serde_json::to_string_pretty(¬ification.params).unwrap(), ); }, ); if let Some(name) = binary.path.file_name() { server.process_name = name.to_string_lossy().into(); } Ok(server) } #[allow(clippy::too_many_arguments)] fn new_internal<Stdin, Stdout, Stderr, F>( server_id: LanguageServerId, server_name: LanguageServerName, stdin: Stdin, stdout: Stdout, stderr: Option<Stderr>, stderr_capture: Arc<Mutex<Option<String>>>, server: Option<Child>, root_path: &Path, working_dir: &Path, code_action_kinds: Option<Vec<CodeActionKind>>, cx: AsyncAppContext, on_unhandled_notification: F, ) -> Self where Stdin: AsyncWrite + Unpin + Send + 'static, Stdout: AsyncRead + Unpin + Send + 'static, Stderr: AsyncRead + Unpin + Send + 'static, F: FnMut(AnyNotification) + 'static + Send + Sync + Clone, { let (outbound_tx, outbound_rx) = channel::unbounded::<String>(); let (output_done_tx, output_done_rx) = barrier::channel(); let notification_handlers = Arc::new(Mutex::new(HashMap::<_, NotificationHandler>::default())); let response_handlers = Arc::new(Mutex::new(Some(HashMap::<_, ResponseHandler>::default()))); let io_handlers = Arc::new(Mutex::new(HashMap::default())); let stdout_input_task = cx.spawn({ let on_unhandled_notification = on_unhandled_notification.clone(); let notification_handlers = notification_handlers.clone(); let response_handlers = response_handlers.clone(); let io_handlers = io_handlers.clone(); move |cx| { Self::handle_input( stdout, on_unhandled_notification, notification_handlers, response_handlers, io_handlers, cx, ) .log_err() } }); let stderr_input_task = stderr .map(|stderr| { let io_handlers = io_handlers.clone(); let stderr_captures = stderr_capture.clone(); cx.spawn(|_| Self::handle_stderr(stderr, io_handlers, stderr_captures).log_err()) }) .unwrap_or_else(|| Task::Ready(Some(None))); let input_task = cx.spawn(|_| async move { let (stdout, stderr) = futures::join!(stdout_input_task, stderr_input_task); stdout.or(stderr) }); let output_task = cx.background_executor().spawn({ Self::handle_output( stdin, outbound_rx, output_done_tx, response_handlers.clone(), io_handlers.clone(), ) .log_err() }); Self { server_id, notification_handlers, response_handlers, io_handlers, name: server_name, process_name: Arc::default(), capabilities: Default::default(), code_action_kinds, next_id: Default::default(), outbound_tx, executor: cx.background_executor().clone(), io_tasks: Mutex::new(Some((input_task, output_task))), output_done_rx: Mutex::new(Some(output_done_rx)), root_path: root_path.to_path_buf(), working_dir: working_dir.to_path_buf(), server: Arc::new(Mutex::new(server)), } } /// List of code action kinds this language server reports being able to emit. pub fn code_action_kinds(&self) -> Option<Vec<CodeActionKind>> { self.code_action_kinds.clone() } async fn handle_input<Stdout, F>( stdout: Stdout, mut on_unhandled_notification: F, notification_handlers: Arc<Mutex<HashMap<&'static str, NotificationHandler>>>, response_handlers: Arc<Mutex<Option<HashMap<RequestId, ResponseHandler>>>>, io_handlers: Arc<Mutex<HashMap<i32, IoHandler>>>, cx: AsyncAppContext, ) -> anyhow::Result<()> where Stdout: AsyncRead + Unpin + Send + 'static, F: FnMut(AnyNotification) + 'static + Send, { use smol::stream::StreamExt; let stdout = BufReader::new(stdout);