1use anyhow::{anyhow, Context, Result};
2use futures::{io::BufWriter, AsyncRead, AsyncWrite};
3use gpui::{executor, Task};
4use parking_lot::{Mutex, RwLock};
5use postage::{barrier, oneshot, prelude::Stream, sink::Sink};
6use serde::{Deserialize, Serialize};
7use serde_json::{json, value::RawValue, Value};
8use smol::{
9 channel,
10 io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader},
11 process::Command,
12};
13use std::{
14 collections::HashMap,
15 future::Future,
16 io::Write,
17 str::FromStr,
18 sync::{
19 atomic::{AtomicBool, AtomicUsize, Ordering::SeqCst},
20 Arc,
21 },
22};
23use std::{path::Path, process::Stdio};
24use util::TryFutureExt;
25
26pub use lsp_types::*;
27
28const JSON_RPC_VERSION: &'static str = "2.0";
29const CONTENT_LEN_HEADER: &'static str = "Content-Length: ";
30
31type NotificationHandler = Box<dyn Send + Sync + Fn(&str)>;
32type ResponseHandler = Box<dyn Send + FnOnce(Result<&str, Error>)>;
33
34pub struct LanguageServer {
35 next_id: AtomicUsize,
36 outbound_tx: RwLock<Option<channel::Sender<Vec<u8>>>>,
37 notification_handlers: Arc<RwLock<HashMap<&'static str, NotificationHandler>>>,
38 response_handlers: Arc<Mutex<HashMap<usize, ResponseHandler>>>,
39 executor: Arc<executor::Background>,
40 io_tasks: Mutex<Option<(Task<Option<()>>, Task<Option<()>>)>>,
41 initialized: barrier::Receiver,
42 output_done_rx: Mutex<Option<barrier::Receiver>>,
43}
44
45pub struct Subscription {
46 method: &'static str,
47 notification_handlers: Arc<RwLock<HashMap<&'static str, NotificationHandler>>>,
48}
49
50#[derive(Serialize, Deserialize)]
51struct Request<'a, T> {
52 jsonrpc: &'a str,
53 id: usize,
54 method: &'a str,
55 params: T,
56}
57
58#[derive(Serialize, Deserialize)]
59struct AnyResponse<'a> {
60 id: usize,
61 #[serde(default)]
62 error: Option<Error>,
63 #[serde(borrow)]
64 result: Option<&'a RawValue>,
65}
66
67#[derive(Serialize, Deserialize)]
68struct Notification<'a, T> {
69 #[serde(borrow)]
70 jsonrpc: &'a str,
71 #[serde(borrow)]
72 method: &'a str,
73 params: T,
74}
75
76#[derive(Deserialize)]
77struct AnyNotification<'a> {
78 #[serde(borrow)]
79 method: &'a str,
80 #[serde(borrow)]
81 params: &'a RawValue,
82}
83
84#[derive(Debug, Serialize, Deserialize)]
85struct Error {
86 message: String,
87}
88
89impl LanguageServer {
90 pub fn new(
91 binary_path: &Path,
92 root_path: &Path,
93 background: Arc<executor::Background>,
94 ) -> Result<Arc<Self>> {
95 let mut server = Command::new(binary_path)
96 .stdin(Stdio::piped())
97 .stdout(Stdio::piped())
98 .stderr(Stdio::inherit())
99 .spawn()?;
100 let stdin = server.stdin.take().unwrap();
101 let stdout = server.stdout.take().unwrap();
102 Self::new_internal(stdin, stdout, root_path, background)
103 }
104
105 fn new_internal<Stdin, Stdout>(
106 stdin: Stdin,
107 stdout: Stdout,
108 root_path: &Path,
109 executor: Arc<executor::Background>,
110 ) -> Result<Arc<Self>>
111 where
112 Stdin: AsyncWrite + Unpin + Send + 'static,
113 Stdout: AsyncRead + Unpin + Send + 'static,
114 {
115 let mut stdin = BufWriter::new(stdin);
116 let mut stdout = BufReader::new(stdout);
117 let (outbound_tx, outbound_rx) = channel::unbounded::<Vec<u8>>();
118 let notification_handlers = Arc::new(RwLock::new(HashMap::<_, NotificationHandler>::new()));
119 let response_handlers = Arc::new(Mutex::new(HashMap::<_, ResponseHandler>::new()));
120 let input_task = executor.spawn(
121 {
122 let notification_handlers = notification_handlers.clone();
123 let response_handlers = response_handlers.clone();
124 async move {
125 let mut buffer = Vec::new();
126 loop {
127 buffer.clear();
128 stdout.read_until(b'\n', &mut buffer).await?;
129 stdout.read_until(b'\n', &mut buffer).await?;
130 let message_len: usize = std::str::from_utf8(&buffer)?
131 .strip_prefix(CONTENT_LEN_HEADER)
132 .ok_or_else(|| anyhow!("invalid header"))?
133 .trim_end()
134 .parse()?;
135
136 buffer.resize(message_len, 0);
137 stdout.read_exact(&mut buffer).await?;
138
139 if let Ok(AnyNotification { method, params }) =
140 serde_json::from_slice(&buffer)
141 {
142 if let Some(handler) = notification_handlers.read().get(method) {
143 handler(params.get());
144 } else {
145 log::info!(
146 "unhandled notification {}:\n{}",
147 method,
148 serde_json::to_string_pretty(
149 &Value::from_str(params.get()).unwrap()
150 )
151 .unwrap()
152 );
153 }
154 } else if let Ok(AnyResponse { id, error, result }) =
155 serde_json::from_slice(&buffer)
156 {
157 if let Some(handler) = response_handlers.lock().remove(&id) {
158 if let Some(error) = error {
159 handler(Err(error));
160 } else if let Some(result) = result {
161 handler(Ok(result.get()));
162 } else {
163 handler(Ok("null"));
164 }
165 }
166 } else {
167 return Err(anyhow!(
168 "failed to deserialize message:\n{}",
169 std::str::from_utf8(&buffer)?
170 ));
171 }
172 }
173 }
174 }
175 .log_err(),
176 );
177 let (output_done_tx, output_done_rx) = barrier::channel();
178 let output_task = executor.spawn(
179 async move {
180 let mut content_len_buffer = Vec::new();
181 while let Ok(message) = outbound_rx.recv().await {
182 content_len_buffer.clear();
183 write!(content_len_buffer, "{}", message.len()).unwrap();
184 stdin.write_all(CONTENT_LEN_HEADER.as_bytes()).await?;
185 stdin.write_all(&content_len_buffer).await?;
186 stdin.write_all("\r\n\r\n".as_bytes()).await?;
187 stdin.write_all(&message).await?;
188 stdin.flush().await?;
189 }
190 drop(output_done_tx);
191 Ok(())
192 }
193 .log_err(),
194 );
195
196 let (initialized_tx, initialized_rx) = barrier::channel();
197 let this = Arc::new(Self {
198 notification_handlers,
199 response_handlers,
200 next_id: Default::default(),
201 outbound_tx: RwLock::new(Some(outbound_tx)),
202 executor: executor.clone(),
203 io_tasks: Mutex::new(Some((input_task, output_task))),
204 initialized: initialized_rx,
205 output_done_rx: Mutex::new(Some(output_done_rx)),
206 });
207
208 let root_uri =
209 lsp_types::Url::from_file_path(root_path).map_err(|_| anyhow!("invalid root path"))?;
210 executor
211 .spawn({
212 let this = this.clone();
213 async move {
214 this.init(root_uri).log_err().await;
215 drop(initialized_tx);
216 }
217 })
218 .detach();
219
220 Ok(this)
221 }
222
223 async fn init(self: Arc<Self>, root_uri: lsp_types::Url) -> Result<()> {
224 #[allow(deprecated)]
225 let params = lsp_types::InitializeParams {
226 process_id: Default::default(),
227 root_path: Default::default(),
228 root_uri: Some(root_uri),
229 initialization_options: Default::default(),
230 capabilities: lsp_types::ClientCapabilities {
231 experimental: Some(json!({
232 "serverStatusNotification": true,
233 })),
234 ..Default::default()
235 },
236 trace: Default::default(),
237 workspace_folders: Default::default(),
238 client_info: Default::default(),
239 locale: Default::default(),
240 };
241
242 let this = self.clone();
243 let request = Self::request_internal::<lsp_types::request::Initialize>(
244 &this.next_id,
245 &this.response_handlers,
246 this.outbound_tx.read().as_ref(),
247 params,
248 );
249 request.await?;
250 Self::notify_internal::<lsp_types::notification::Initialized>(
251 this.outbound_tx.read().as_ref(),
252 lsp_types::InitializedParams {},
253 )?;
254 Ok(())
255 }
256
257 pub fn shutdown(&self) -> Option<impl 'static + Send + Future<Output = Result<()>>> {
258 if let Some(tasks) = self.io_tasks.lock().take() {
259 let response_handlers = self.response_handlers.clone();
260 let outbound_tx = self.outbound_tx.write().take();
261 let next_id = AtomicUsize::new(self.next_id.load(SeqCst));
262 let mut output_done = self.output_done_rx.lock().take().unwrap();
263 Some(async move {
264 Self::request_internal::<lsp_types::request::Shutdown>(
265 &next_id,
266 &response_handlers,
267 outbound_tx.as_ref(),
268 (),
269 )
270 .await?;
271 Self::notify_internal::<lsp_types::notification::Exit>(outbound_tx.as_ref(), ())?;
272 drop(outbound_tx);
273 output_done.recv().await;
274 drop(tasks);
275 Ok(())
276 })
277 } else {
278 None
279 }
280 }
281
282 pub fn on_notification<T, F>(&self, f: F) -> Subscription
283 where
284 T: lsp_types::notification::Notification,
285 F: 'static + Send + Sync + Fn(T::Params),
286 {
287 let prev_handler = self.notification_handlers.write().insert(
288 T::METHOD,
289 Box::new(
290 move |notification| match serde_json::from_str(notification) {
291 Ok(notification) => f(notification),
292 Err(err) => log::error!("error parsing notification {}: {}", T::METHOD, err),
293 },
294 ),
295 );
296
297 assert!(
298 prev_handler.is_none(),
299 "registered multiple handlers for the same notification"
300 );
301
302 Subscription {
303 method: T::METHOD,
304 notification_handlers: self.notification_handlers.clone(),
305 }
306 }
307
308 pub fn request<T: lsp_types::request::Request>(
309 self: Arc<Self>,
310 params: T::Params,
311 ) -> impl Future<Output = Result<T::Result>>
312 where
313 T::Result: 'static + Send,
314 {
315 let this = self.clone();
316 async move {
317 this.initialized.clone().recv().await;
318 Self::request_internal::<T>(
319 &this.next_id,
320 &this.response_handlers,
321 this.outbound_tx.read().as_ref(),
322 params,
323 )
324 .await
325 }
326 }
327
328 fn request_internal<T: lsp_types::request::Request>(
329 next_id: &AtomicUsize,
330 response_handlers: &Mutex<HashMap<usize, ResponseHandler>>,
331 outbound_tx: Option<&channel::Sender<Vec<u8>>>,
332 params: T::Params,
333 ) -> impl 'static + Future<Output = Result<T::Result>>
334 where
335 T::Result: 'static + Send,
336 {
337 let id = next_id.fetch_add(1, SeqCst);
338 let message = serde_json::to_vec(&Request {
339 jsonrpc: JSON_RPC_VERSION,
340 id,
341 method: T::METHOD,
342 params,
343 })
344 .unwrap();
345 let mut response_handlers = response_handlers.lock();
346 let (mut tx, mut rx) = oneshot::channel();
347 response_handlers.insert(
348 id,
349 Box::new(move |result| {
350 let response = match result {
351 Ok(response) => {
352 serde_json::from_str(response).context("failed to deserialize response")
353 }
354 Err(error) => Err(anyhow!("{}", error.message)),
355 };
356 let _ = tx.try_send(response);
357 }),
358 );
359
360 let send = outbound_tx
361 .as_ref()
362 .ok_or_else(|| {
363 anyhow!("tried to send a request to a language server that has been shut down")
364 })
365 .and_then(|outbound_tx| {
366 outbound_tx.try_send(message)?;
367 Ok(())
368 });
369 async move {
370 send?;
371 rx.recv().await.unwrap()
372 }
373 }
374
375 pub fn notify<T: lsp_types::notification::Notification>(
376 self: &Arc<Self>,
377 params: T::Params,
378 ) -> impl Future<Output = Result<()>> {
379 let this = self.clone();
380 async move {
381 this.initialized.clone().recv().await;
382 Self::notify_internal::<T>(this.outbound_tx.read().as_ref(), params)?;
383 Ok(())
384 }
385 }
386
387 fn notify_internal<T: lsp_types::notification::Notification>(
388 outbound_tx: Option<&channel::Sender<Vec<u8>>>,
389 params: T::Params,
390 ) -> Result<()> {
391 let message = serde_json::to_vec(&Notification {
392 jsonrpc: JSON_RPC_VERSION,
393 method: T::METHOD,
394 params,
395 })
396 .unwrap();
397 let outbound_tx = outbound_tx
398 .as_ref()
399 .ok_or_else(|| anyhow!("tried to notify a language server that has been shut down"))?;
400 outbound_tx.try_send(message)?;
401 Ok(())
402 }
403}
404
405impl Drop for LanguageServer {
406 fn drop(&mut self) {
407 if let Some(shutdown) = self.shutdown() {
408 self.executor.spawn(shutdown).detach();
409 }
410 }
411}
412
413impl Subscription {
414 pub fn detach(mut self) {
415 self.method = "";
416 }
417}
418
419impl Drop for Subscription {
420 fn drop(&mut self) {
421 self.notification_handlers.write().remove(self.method);
422 }
423}
424
425#[cfg(any(test, feature = "test-support"))]
426pub struct FakeLanguageServer {
427 buffer: Vec<u8>,
428 stdin: smol::io::BufReader<async_pipe::PipeReader>,
429 stdout: smol::io::BufWriter<async_pipe::PipeWriter>,
430 pub started: Arc<AtomicBool>,
431}
432
433#[cfg(any(test, feature = "test-support"))]
434pub struct RequestId<T> {
435 id: usize,
436 _type: std::marker::PhantomData<T>,
437}
438
439#[cfg(any(test, feature = "test-support"))]
440impl LanguageServer {
441 pub async fn fake(executor: Arc<executor::Background>) -> (Arc<Self>, FakeLanguageServer) {
442 let stdin = async_pipe::pipe();
443 let stdout = async_pipe::pipe();
444 let mut fake = FakeLanguageServer {
445 stdin: smol::io::BufReader::new(stdin.1),
446 stdout: smol::io::BufWriter::new(stdout.0),
447 buffer: Vec::new(),
448 started: Arc::new(AtomicBool::new(true)),
449 };
450
451 let server = Self::new_internal(stdin.0, stdout.1, Path::new("/"), executor).unwrap();
452
453 let (init_id, _) = fake.receive_request::<request::Initialize>().await;
454 fake.respond(init_id, InitializeResult::default()).await;
455 fake.receive_notification::<notification::Initialized>()
456 .await;
457
458 (server, fake)
459 }
460}
461
462#[cfg(any(test, feature = "test-support"))]
463impl FakeLanguageServer {
464 pub async fn notify<T: notification::Notification>(&mut self, params: T::Params) {
465 if !self.started.load(std::sync::atomic::Ordering::SeqCst) {
466 panic!("can't simulate an LSP notification before the server has been started");
467 }
468 let message = serde_json::to_vec(&Notification {
469 jsonrpc: JSON_RPC_VERSION,
470 method: T::METHOD,
471 params,
472 })
473 .unwrap();
474 self.send(message).await;
475 }
476
477 pub async fn respond<'a, T: request::Request>(
478 &mut self,
479 request_id: RequestId<T>,
480 result: T::Result,
481 ) {
482 let result = serde_json::to_string(&result).unwrap();
483 let message = serde_json::to_vec(&AnyResponse {
484 id: request_id.id,
485 error: None,
486 result: Some(&RawValue::from_string(result).unwrap()),
487 })
488 .unwrap();
489 self.send(message).await;
490 }
491
492 pub async fn receive_request<T: request::Request>(&mut self) -> (RequestId<T>, T::Params) {
493 self.receive().await;
494 let request = serde_json::from_slice::<Request<T::Params>>(&self.buffer).unwrap();
495 assert_eq!(request.method, T::METHOD);
496 assert_eq!(request.jsonrpc, JSON_RPC_VERSION);
497 (
498 RequestId {
499 id: request.id,
500 _type: std::marker::PhantomData,
501 },
502 request.params,
503 )
504 }
505
506 pub async fn receive_notification<T: notification::Notification>(&mut self) -> T::Params {
507 self.receive().await;
508 let notification = serde_json::from_slice::<Notification<T::Params>>(&self.buffer).unwrap();
509 assert_eq!(notification.method, T::METHOD);
510 notification.params
511 }
512
513 async fn send(&mut self, message: Vec<u8>) {
514 self.stdout
515 .write_all(CONTENT_LEN_HEADER.as_bytes())
516 .await
517 .unwrap();
518 self.stdout
519 .write_all((format!("{}", message.len())).as_bytes())
520 .await
521 .unwrap();
522 self.stdout.write_all("\r\n\r\n".as_bytes()).await.unwrap();
523 self.stdout.write_all(&message).await.unwrap();
524 self.stdout.flush().await.unwrap();
525 }
526
527 async fn receive(&mut self) {
528 self.buffer.clear();
529 self.stdin
530 .read_until(b'\n', &mut self.buffer)
531 .await
532 .unwrap();
533 self.stdin
534 .read_until(b'\n', &mut self.buffer)
535 .await
536 .unwrap();
537 let message_len: usize = std::str::from_utf8(&self.buffer)
538 .unwrap()
539 .strip_prefix(CONTENT_LEN_HEADER)
540 .unwrap()
541 .trim_end()
542 .parse()
543 .unwrap();
544 self.buffer.resize(message_len, 0);
545 self.stdin.read_exact(&mut self.buffer).await.unwrap();
546 }
547}
548
549#[cfg(test)]
550mod tests {
551 use super::*;
552 use gpui::TestAppContext;
553 use simplelog::SimpleLogger;
554 use unindent::Unindent;
555 use util::test::temp_tree;
556
557 #[gpui::test]
558 async fn test_basic(cx: TestAppContext) {
559 let lib_source = r#"
560 fn fun() {
561 let hello = "world";
562 }
563 "#
564 .unindent();
565 let root_dir = temp_tree(json!({
566 "Cargo.toml": r#"
567 [package]
568 name = "temp"
569 version = "0.1.0"
570 edition = "2018"
571 "#.unindent(),
572 "src": {
573 "lib.rs": &lib_source
574 }
575 }));
576 let lib_file_uri =
577 lsp_types::Url::from_file_path(root_dir.path().join("src/lib.rs")).unwrap();
578
579 let server = cx.read(|cx| {
580 LanguageServer::new(
581 Path::new("rust-analyzer"),
582 root_dir.path(),
583 cx.background().clone(),
584 )
585 .unwrap()
586 });
587 server.next_idle_notification().await;
588
589 server
590 .notify::<lsp_types::notification::DidOpenTextDocument>(
591 lsp_types::DidOpenTextDocumentParams {
592 text_document: lsp_types::TextDocumentItem::new(
593 lib_file_uri.clone(),
594 "rust".to_string(),
595 0,
596 lib_source,
597 ),
598 },
599 )
600 .await
601 .unwrap();
602
603 let hover = server
604 .request::<lsp_types::request::HoverRequest>(lsp_types::HoverParams {
605 text_document_position_params: lsp_types::TextDocumentPositionParams {
606 text_document: lsp_types::TextDocumentIdentifier::new(lib_file_uri),
607 position: lsp_types::Position::new(1, 21),
608 },
609 work_done_progress_params: Default::default(),
610 })
611 .await
612 .unwrap()
613 .unwrap();
614 assert_eq!(
615 hover.contents,
616 lsp_types::HoverContents::Markup(lsp_types::MarkupContent {
617 kind: lsp_types::MarkupKind::Markdown,
618 value: "&str".to_string()
619 })
620 );
621 }
622
623 #[gpui::test]
624 async fn test_fake(cx: TestAppContext) {
625 SimpleLogger::init(log::LevelFilter::Info, Default::default()).unwrap();
626
627 let (server, mut fake) = LanguageServer::fake(cx.background()).await;
628
629 let (message_tx, message_rx) = channel::unbounded();
630 let (diagnostics_tx, diagnostics_rx) = channel::unbounded();
631 server
632 .on_notification::<notification::ShowMessage, _>(move |params| {
633 message_tx.try_send(params).unwrap()
634 })
635 .detach();
636 server
637 .on_notification::<notification::PublishDiagnostics, _>(move |params| {
638 diagnostics_tx.try_send(params).unwrap()
639 })
640 .detach();
641
642 server
643 .notify::<notification::DidOpenTextDocument>(DidOpenTextDocumentParams {
644 text_document: TextDocumentItem::new(
645 Url::from_str("file://a/b").unwrap(),
646 "rust".to_string(),
647 0,
648 "".to_string(),
649 ),
650 })
651 .await
652 .unwrap();
653 assert_eq!(
654 fake.receive_notification::<notification::DidOpenTextDocument>()
655 .await
656 .text_document
657 .uri
658 .as_str(),
659 "file://a/b"
660 );
661
662 fake.notify::<notification::ShowMessage>(ShowMessageParams {
663 typ: MessageType::ERROR,
664 message: "ok".to_string(),
665 })
666 .await;
667 fake.notify::<notification::PublishDiagnostics>(PublishDiagnosticsParams {
668 uri: Url::from_str("file://b/c").unwrap(),
669 version: Some(5),
670 diagnostics: vec![],
671 })
672 .await;
673 assert_eq!(message_rx.recv().await.unwrap().message, "ok");
674 assert_eq!(
675 diagnostics_rx.recv().await.unwrap().uri.as_str(),
676 "file://b/c"
677 );
678
679 drop(server);
680 let (shutdown_request, _) = fake.receive_request::<lsp_types::request::Shutdown>().await;
681 fake.respond(shutdown_request, ()).await;
682 fake.receive_notification::<lsp_types::notification::Exit>()
683 .await;
684 }
685
686 impl LanguageServer {
687 async fn next_idle_notification(self: &Arc<Self>) {
688 let (tx, rx) = channel::unbounded();
689 let _subscription =
690 self.on_notification::<ServerStatusNotification, _>(move |params| {
691 if params.quiescent {
692 tx.try_send(()).unwrap();
693 }
694 });
695 let _ = rx.recv().await;
696 }
697 }
698
699 pub enum ServerStatusNotification {}
700
701 impl lsp_types::notification::Notification for ServerStatusNotification {
702 type Params = ServerStatusParams;
703 const METHOD: &'static str = "experimental/serverStatus";
704 }
705
706 #[derive(Deserialize, Serialize, PartialEq, Eq, Clone)]
707 pub struct ServerStatusParams {
708 pub quiescent: bool,
709 }
710}