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