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