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::{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}
431
432#[cfg(any(test, feature = "test-support"))]
433pub struct RequestId<T> {
434 id: usize,
435 _type: std::marker::PhantomData<T>,
436}
437
438#[cfg(any(test, feature = "test-support"))]
439impl LanguageServer {
440 pub async fn fake(executor: Arc<executor::Background>) -> (Arc<Self>, FakeLanguageServer) {
441 let stdin = async_pipe::pipe();
442 let stdout = async_pipe::pipe();
443 let mut fake = FakeLanguageServer {
444 stdin: smol::io::BufReader::new(stdin.1),
445 stdout: smol::io::BufWriter::new(stdout.0),
446 buffer: Vec::new(),
447 };
448
449 let server = Self::new_internal(stdin.0, stdout.1, Path::new("/"), executor).unwrap();
450
451 let (init_id, _) = fake.receive_request::<request::Initialize>().await;
452 fake.respond(init_id, InitializeResult::default()).await;
453 fake.receive_notification::<notification::Initialized>()
454 .await;
455
456 (server, fake)
457 }
458}
459
460#[cfg(any(test, feature = "test-support"))]
461impl FakeLanguageServer {
462 pub async fn notify<T: notification::Notification>(&mut self, params: T::Params) {
463 let message = serde_json::to_vec(&Notification {
464 jsonrpc: JSON_RPC_VERSION,
465 method: T::METHOD,
466 params,
467 })
468 .unwrap();
469 self.send(message).await;
470 }
471
472 pub async fn respond<'a, T: request::Request>(
473 &mut self,
474 request_id: RequestId<T>,
475 result: T::Result,
476 ) {
477 let result = serde_json::to_string(&result).unwrap();
478 let message = serde_json::to_vec(&AnyResponse {
479 id: request_id.id,
480 error: None,
481 result: Some(&RawValue::from_string(result).unwrap()),
482 })
483 .unwrap();
484 self.send(message).await;
485 }
486
487 pub async fn receive_request<T: request::Request>(&mut self) -> (RequestId<T>, T::Params) {
488 self.receive().await;
489 let request = serde_json::from_slice::<Request<T::Params>>(&self.buffer).unwrap();
490 assert_eq!(request.method, T::METHOD);
491 assert_eq!(request.jsonrpc, JSON_RPC_VERSION);
492 (
493 RequestId {
494 id: request.id,
495 _type: std::marker::PhantomData,
496 },
497 request.params,
498 )
499 }
500
501 pub async fn receive_notification<T: notification::Notification>(&mut self) -> T::Params {
502 self.receive().await;
503 let notification = serde_json::from_slice::<Notification<T::Params>>(&self.buffer).unwrap();
504 assert_eq!(notification.method, T::METHOD);
505 notification.params
506 }
507
508 async fn send(&mut self, message: Vec<u8>) {
509 self.stdout
510 .write_all(CONTENT_LEN_HEADER.as_bytes())
511 .await
512 .unwrap();
513 self.stdout
514 .write_all((format!("{}", message.len())).as_bytes())
515 .await
516 .unwrap();
517 self.stdout.write_all("\r\n\r\n".as_bytes()).await.unwrap();
518 self.stdout.write_all(&message).await.unwrap();
519 self.stdout.flush().await.unwrap();
520 }
521
522 async fn receive(&mut self) {
523 self.buffer.clear();
524 self.stdin
525 .read_until(b'\n', &mut self.buffer)
526 .await
527 .unwrap();
528 self.stdin
529 .read_until(b'\n', &mut self.buffer)
530 .await
531 .unwrap();
532 let message_len: usize = std::str::from_utf8(&self.buffer)
533 .unwrap()
534 .strip_prefix(CONTENT_LEN_HEADER)
535 .unwrap()
536 .trim_end()
537 .parse()
538 .unwrap();
539 self.buffer.resize(message_len, 0);
540 self.stdin.read_exact(&mut self.buffer).await.unwrap();
541 }
542}
543
544#[cfg(test)]
545mod tests {
546 use super::*;
547 use gpui::TestAppContext;
548 use simplelog::SimpleLogger;
549 use unindent::Unindent;
550 use util::test::temp_tree;
551
552 #[gpui::test]
553 async fn test_basic(cx: TestAppContext) {
554 let lib_source = r#"
555 fn fun() {
556 let hello = "world";
557 }
558 "#
559 .unindent();
560 let root_dir = temp_tree(json!({
561 "Cargo.toml": r#"
562 [package]
563 name = "temp"
564 version = "0.1.0"
565 edition = "2018"
566 "#.unindent(),
567 "src": {
568 "lib.rs": &lib_source
569 }
570 }));
571 let lib_file_uri =
572 lsp_types::Url::from_file_path(root_dir.path().join("src/lib.rs")).unwrap();
573
574 let server = cx.read(|cx| {
575 LanguageServer::new(
576 Path::new("rust-analyzer"),
577 root_dir.path(),
578 cx.background().clone(),
579 )
580 .unwrap()
581 });
582 server.next_idle_notification().await;
583
584 server
585 .notify::<lsp_types::notification::DidOpenTextDocument>(
586 lsp_types::DidOpenTextDocumentParams {
587 text_document: lsp_types::TextDocumentItem::new(
588 lib_file_uri.clone(),
589 "rust".to_string(),
590 0,
591 lib_source,
592 ),
593 },
594 )
595 .await
596 .unwrap();
597
598 let hover = server
599 .request::<lsp_types::request::HoverRequest>(lsp_types::HoverParams {
600 text_document_position_params: lsp_types::TextDocumentPositionParams {
601 text_document: lsp_types::TextDocumentIdentifier::new(lib_file_uri),
602 position: lsp_types::Position::new(1, 21),
603 },
604 work_done_progress_params: Default::default(),
605 })
606 .await
607 .unwrap()
608 .unwrap();
609 assert_eq!(
610 hover.contents,
611 lsp_types::HoverContents::Markup(lsp_types::MarkupContent {
612 kind: lsp_types::MarkupKind::Markdown,
613 value: "&str".to_string()
614 })
615 );
616 }
617
618 #[gpui::test]
619 async fn test_fake(cx: TestAppContext) {
620 SimpleLogger::init(log::LevelFilter::Info, Default::default()).unwrap();
621
622 let (server, mut fake) = LanguageServer::fake(cx.background()).await;
623
624 let (message_tx, message_rx) = channel::unbounded();
625 let (diagnostics_tx, diagnostics_rx) = channel::unbounded();
626 server
627 .on_notification::<notification::ShowMessage, _>(move |params| {
628 message_tx.try_send(params).unwrap()
629 })
630 .detach();
631 server
632 .on_notification::<notification::PublishDiagnostics, _>(move |params| {
633 diagnostics_tx.try_send(params).unwrap()
634 })
635 .detach();
636
637 server
638 .notify::<notification::DidOpenTextDocument>(DidOpenTextDocumentParams {
639 text_document: TextDocumentItem::new(
640 Url::from_str("file://a/b").unwrap(),
641 "rust".to_string(),
642 0,
643 "".to_string(),
644 ),
645 })
646 .await
647 .unwrap();
648 assert_eq!(
649 fake.receive_notification::<notification::DidOpenTextDocument>()
650 .await
651 .text_document
652 .uri
653 .as_str(),
654 "file://a/b"
655 );
656
657 fake.notify::<notification::ShowMessage>(ShowMessageParams {
658 typ: MessageType::ERROR,
659 message: "ok".to_string(),
660 })
661 .await;
662 fake.notify::<notification::PublishDiagnostics>(PublishDiagnosticsParams {
663 uri: Url::from_str("file://b/c").unwrap(),
664 version: Some(5),
665 diagnostics: vec![],
666 })
667 .await;
668 assert_eq!(message_rx.recv().await.unwrap().message, "ok");
669 assert_eq!(
670 diagnostics_rx.recv().await.unwrap().uri.as_str(),
671 "file://b/c"
672 );
673
674 drop(server);
675 let (shutdown_request, _) = fake.receive_request::<lsp_types::request::Shutdown>().await;
676 fake.respond(shutdown_request, ()).await;
677 fake.receive_notification::<lsp_types::notification::Exit>()
678 .await;
679 }
680
681 impl LanguageServer {
682 async fn next_idle_notification(self: &Arc<Self>) {
683 let (tx, rx) = channel::unbounded();
684 let _subscription =
685 self.on_notification::<ServerStatusNotification, _>(move |params| {
686 if params.quiescent {
687 tx.try_send(()).unwrap();
688 }
689 });
690 let _ = rx.recv().await;
691 }
692 }
693
694 pub enum ServerStatusNotification {}
695
696 impl lsp_types::notification::Notification for ServerStatusNotification {
697 type Params = ServerStatusParams;
698 const METHOD: &'static str = "experimental/serverStatus";
699 }
700
701 #[derive(Deserialize, Serialize, PartialEq, Eq, Clone)]
702 pub struct ServerStatusParams {
703 pub quiescent: bool,
704 }
705}