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