lsp.rs

  1pub use lsp_types::*;
  2
  3use anyhow::{anyhow, Context, Result};
  4use collections::HashMap;
  5use futures::{channel::oneshot, io::BufWriter, AsyncRead, AsyncWrite};
  6use gpui::{executor, AsyncAppContext, Task};
  7use parking_lot::Mutex;
  8use postage::{barrier, prelude::Stream};
  9use serde::{de::DeserializeOwned, Deserialize, Serialize};
 10use serde_json::{json, value::RawValue, Value};
 11use smol::{
 12    channel,
 13    io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader},
 14    process,
 15};
 16use std::{
 17    future::Future,
 18    io::Write,
 19    path::PathBuf,
 20    str::FromStr,
 21    sync::{
 22        atomic::{AtomicUsize, Ordering::SeqCst},
 23        Arc,
 24    },
 25};
 26use std::{path::Path, process::Stdio};
 27use util::{ResultExt, TryFutureExt};
 28
 29const JSON_RPC_VERSION: &'static str = "2.0";
 30const CONTENT_LEN_HEADER: &'static str = "Content-Length: ";
 31
 32type NotificationHandler = Box<dyn Send + FnMut(Option<usize>, &str, AsyncAppContext)>;
 33type ResponseHandler = Box<dyn Send + FnOnce(Result<&str, Error>)>;
 34
 35pub struct LanguageServer {
 36    server_id: usize,
 37    next_id: AtomicUsize,
 38    outbound_tx: channel::Sender<Vec<u8>>,
 39    name: String,
 40    capabilities: ServerCapabilities,
 41    notification_handlers: Arc<Mutex<HashMap<&'static str, NotificationHandler>>>,
 42    response_handlers: Arc<Mutex<HashMap<usize, ResponseHandler>>>,
 43    executor: Arc<executor::Background>,
 44    io_tasks: Mutex<Option<(Task<Option<()>>, Task<Option<()>>)>>,
 45    output_done_rx: Mutex<Option<barrier::Receiver>>,
 46    root_path: PathBuf,
 47}
 48
 49pub struct Subscription {
 50    method: &'static str,
 51    notification_handlers: Arc<Mutex<HashMap<&'static str, NotificationHandler>>>,
 52}
 53
 54#[derive(Serialize, Deserialize)]
 55struct Request<'a, T> {
 56    jsonrpc: &'a str,
 57    id: usize,
 58    method: &'a str,
 59    params: T,
 60}
 61
 62#[derive(Serialize, Deserialize)]
 63struct AnyResponse<'a> {
 64    id: usize,
 65    #[serde(default)]
 66    error: Option<Error>,
 67    #[serde(borrow)]
 68    result: Option<&'a RawValue>,
 69}
 70
 71#[derive(Serialize)]
 72struct Response<T> {
 73    id: usize,
 74    result: Option<T>,
 75    error: Option<Error>,
 76}
 77
 78#[derive(Serialize, Deserialize)]
 79struct Notification<'a, T> {
 80    #[serde(borrow)]
 81    jsonrpc: &'a str,
 82    #[serde(borrow)]
 83    method: &'a str,
 84    params: T,
 85}
 86
 87#[derive(Deserialize)]
 88struct AnyNotification<'a> {
 89    #[serde(default)]
 90    id: Option<usize>,
 91    #[serde(borrow)]
 92    method: &'a str,
 93    #[serde(borrow)]
 94    params: &'a RawValue,
 95}
 96
 97#[derive(Debug, Serialize, Deserialize)]
 98struct Error {
 99    message: String,
100}
101
102impl LanguageServer {
103    pub fn new(
104        server_id: usize,
105        binary_path: &Path,
106        args: &[&str],
107        root_path: &Path,
108        cx: AsyncAppContext,
109    ) -> Result<Self> {
110        let working_dir = if root_path.is_dir() {
111            root_path
112        } else {
113            root_path.parent().unwrap_or(Path::new("/"))
114        };
115        let mut server = process::Command::new(binary_path)
116            .current_dir(working_dir)
117            .args(args)
118            .stdin(Stdio::piped())
119            .stdout(Stdio::piped())
120            .stderr(Stdio::inherit())
121            .spawn()?;
122        let stdin = server.stdin.take().unwrap();
123        let stdout = server.stdout.take().unwrap();
124        let mut server =
125            Self::new_internal(server_id, stdin, stdout, root_path, cx, |notification| {
126                log::info!(
127                    "unhandled notification {}:\n{}",
128                    notification.method,
129                    serde_json::to_string_pretty(
130                        &Value::from_str(notification.params.get()).unwrap()
131                    )
132                    .unwrap()
133                );
134            });
135        if let Some(name) = binary_path.file_name() {
136            server.name = name.to_string_lossy().to_string();
137        }
138        Ok(server)
139    }
140
141    fn new_internal<Stdin, Stdout, F>(
142        server_id: usize,
143        stdin: Stdin,
144        stdout: Stdout,
145        root_path: &Path,
146        cx: AsyncAppContext,
147        mut on_unhandled_notification: F,
148    ) -> Self
149    where
150        Stdin: AsyncWrite + Unpin + Send + 'static,
151        Stdout: AsyncRead + Unpin + Send + 'static,
152        F: FnMut(AnyNotification) + 'static + Send,
153    {
154        let mut stdin = BufWriter::new(stdin);
155        let mut stdout = BufReader::new(stdout);
156        let (outbound_tx, outbound_rx) = channel::unbounded::<Vec<u8>>();
157        let notification_handlers =
158            Arc::new(Mutex::new(HashMap::<_, NotificationHandler>::default()));
159        let response_handlers = Arc::new(Mutex::new(HashMap::<_, ResponseHandler>::default()));
160        let input_task = cx.spawn(|cx| {
161            let notification_handlers = notification_handlers.clone();
162            let response_handlers = response_handlers.clone();
163            async move {
164                let _clear_response_handlers = ClearResponseHandlers(response_handlers.clone());
165                let mut buffer = Vec::new();
166                loop {
167                    buffer.clear();
168                    stdout.read_until(b'\n', &mut buffer).await?;
169                    stdout.read_until(b'\n', &mut buffer).await?;
170                    let message_len: usize = std::str::from_utf8(&buffer)?
171                        .strip_prefix(CONTENT_LEN_HEADER)
172                        .ok_or_else(|| anyhow!("invalid header"))?
173                        .trim_end()
174                        .parse()?;
175
176                    buffer.resize(message_len, 0);
177                    stdout.read_exact(&mut buffer).await?;
178                    log::trace!("incoming message:{}", String::from_utf8_lossy(&buffer));
179
180                    if let Ok(msg) = serde_json::from_slice::<AnyNotification>(&buffer) {
181                        if let Some(handler) = notification_handlers.lock().get_mut(msg.method) {
182                            handler(msg.id, msg.params.get(), cx.clone());
183                        } else {
184                            on_unhandled_notification(msg);
185                        }
186                    } else if let Ok(AnyResponse { id, error, result }) =
187                        serde_json::from_slice(&buffer)
188                    {
189                        if let Some(handler) = response_handlers.lock().remove(&id) {
190                            if let Some(error) = error {
191                                handler(Err(error));
192                            } else if let Some(result) = result {
193                                handler(Ok(result.get()));
194                            } else {
195                                handler(Ok("null"));
196                            }
197                        }
198                    } else {
199                        return Err(anyhow!(
200                            "failed to deserialize message:\n{}",
201                            std::str::from_utf8(&buffer)?
202                        ));
203                    }
204                }
205            }
206            .log_err()
207        });
208        let (output_done_tx, output_done_rx) = barrier::channel();
209        let output_task = cx.background().spawn({
210            let response_handlers = response_handlers.clone();
211            async move {
212                let _clear_response_handlers = ClearResponseHandlers(response_handlers);
213                let mut content_len_buffer = Vec::new();
214                while let Ok(message) = outbound_rx.recv().await {
215                    log::trace!("outgoing message:{}", String::from_utf8_lossy(&message));
216                    content_len_buffer.clear();
217                    write!(content_len_buffer, "{}", message.len()).unwrap();
218                    stdin.write_all(CONTENT_LEN_HEADER.as_bytes()).await?;
219                    stdin.write_all(&content_len_buffer).await?;
220                    stdin.write_all("\r\n\r\n".as_bytes()).await?;
221                    stdin.write_all(&message).await?;
222                    stdin.flush().await?;
223                }
224                drop(output_done_tx);
225                Ok(())
226            }
227            .log_err()
228        });
229
230        Self {
231            server_id,
232            notification_handlers,
233            response_handlers,
234            name: Default::default(),
235            capabilities: Default::default(),
236            next_id: Default::default(),
237            outbound_tx,
238            executor: cx.background().clone(),
239            io_tasks: Mutex::new(Some((input_task, output_task))),
240            output_done_rx: Mutex::new(Some(output_done_rx)),
241            root_path: root_path.to_path_buf(),
242        }
243    }
244
245    pub async fn initialize(mut self, options: Option<Value>) -> Result<Arc<Self>> {
246        let root_uri = Url::from_file_path(&self.root_path).unwrap();
247        #[allow(deprecated)]
248        let params = InitializeParams {
249            process_id: Default::default(),
250            root_path: Default::default(),
251            root_uri: Some(root_uri),
252            initialization_options: options,
253            capabilities: ClientCapabilities {
254                workspace: Some(WorkspaceClientCapabilities {
255                    configuration: Some(true),
256                    did_change_configuration: Some(DynamicRegistrationClientCapabilities {
257                        dynamic_registration: Some(true),
258                    }),
259                    ..Default::default()
260                }),
261                text_document: Some(TextDocumentClientCapabilities {
262                    definition: Some(GotoCapability {
263                        link_support: Some(true),
264                        ..Default::default()
265                    }),
266                    code_action: Some(CodeActionClientCapabilities {
267                        code_action_literal_support: Some(CodeActionLiteralSupport {
268                            code_action_kind: CodeActionKindLiteralSupport {
269                                value_set: vec![
270                                    CodeActionKind::REFACTOR.as_str().into(),
271                                    CodeActionKind::QUICKFIX.as_str().into(),
272                                    CodeActionKind::SOURCE.as_str().into(),
273                                ],
274                            },
275                        }),
276                        data_support: Some(true),
277                        resolve_support: Some(CodeActionCapabilityResolveSupport {
278                            properties: vec!["edit".to_string(), "command".to_string()],
279                        }),
280                        ..Default::default()
281                    }),
282                    completion: Some(CompletionClientCapabilities {
283                        completion_item: Some(CompletionItemCapability {
284                            snippet_support: Some(true),
285                            resolve_support: Some(CompletionItemCapabilityResolveSupport {
286                                properties: vec!["additionalTextEdits".to_string()],
287                            }),
288                            ..Default::default()
289                        }),
290                        ..Default::default()
291                    }),
292                    ..Default::default()
293                }),
294                experimental: Some(json!({
295                    "serverStatusNotification": true,
296                })),
297                window: Some(WindowClientCapabilities {
298                    work_done_progress: Some(true),
299                    ..Default::default()
300                }),
301                ..Default::default()
302            },
303            trace: Default::default(),
304            workspace_folders: Default::default(),
305            client_info: Default::default(),
306            locale: Default::default(),
307        };
308
309        let response = self.request::<request::Initialize>(params).await?;
310        if let Some(info) = response.server_info {
311            self.name = info.name;
312        }
313        self.capabilities = response.capabilities;
314
315        self.notify::<notification::Initialized>(InitializedParams {})?;
316        Ok(Arc::new(self))
317    }
318
319    pub fn shutdown(&self) -> Option<impl 'static + Send + Future<Output = Option<()>>> {
320        if let Some(tasks) = self.io_tasks.lock().take() {
321            let response_handlers = self.response_handlers.clone();
322            let next_id = AtomicUsize::new(self.next_id.load(SeqCst));
323            let outbound_tx = self.outbound_tx.clone();
324            let mut output_done = self.output_done_rx.lock().take().unwrap();
325            let shutdown_request = Self::request_internal::<request::Shutdown>(
326                &next_id,
327                &response_handlers,
328                &outbound_tx,
329                (),
330            );
331            let exit = Self::notify_internal::<notification::Exit>(&outbound_tx, ());
332            outbound_tx.close();
333            Some(
334                async move {
335                    log::debug!("language server shutdown started");
336                    shutdown_request.await?;
337                    response_handlers.lock().clear();
338                    exit?;
339                    output_done.recv().await;
340                    log::debug!("language server shutdown finished");
341                    drop(tasks);
342                    Ok(())
343                }
344                .log_err(),
345            )
346        } else {
347            None
348        }
349    }
350
351    #[must_use]
352    pub fn on_notification<T, F>(&self, f: F) -> Subscription
353    where
354        T: notification::Notification,
355        F: 'static + Send + FnMut(T::Params, AsyncAppContext),
356    {
357        self.on_custom_notification(T::METHOD, f)
358    }
359
360    #[must_use]
361    pub fn on_request<T, F, Fut>(&self, f: F) -> Subscription
362    where
363        T: request::Request,
364        T::Params: 'static + Send,
365        F: 'static + Send + FnMut(T::Params, AsyncAppContext) -> Fut,
366        Fut: 'static + Future<Output = Result<T::Result>>,
367    {
368        self.on_custom_request(T::METHOD, f)
369    }
370
371    pub fn remove_request_handler<T: request::Request>(&self) {
372        self.notification_handlers.lock().remove(T::METHOD);
373    }
374
375    #[must_use]
376    pub fn on_custom_notification<Params, F>(&self, method: &'static str, mut f: F) -> Subscription
377    where
378        F: 'static + Send + FnMut(Params, AsyncAppContext),
379        Params: DeserializeOwned,
380    {
381        let prev_handler = self.notification_handlers.lock().insert(
382            method,
383            Box::new(move |_, params, cx| {
384                if let Some(params) = serde_json::from_str(params).log_err() {
385                    f(params, cx);
386                }
387            }),
388        );
389        assert!(
390            prev_handler.is_none(),
391            "registered multiple handlers for the same LSP method"
392        );
393        Subscription {
394            method,
395            notification_handlers: self.notification_handlers.clone(),
396        }
397    }
398
399    #[must_use]
400    pub fn on_custom_request<Params, Res, Fut, F>(
401        &self,
402        method: &'static str,
403        mut f: F,
404    ) -> Subscription
405    where
406        F: 'static + Send + FnMut(Params, AsyncAppContext) -> Fut,
407        Fut: 'static + Future<Output = Result<Res>>,
408        Params: DeserializeOwned + Send + 'static,
409        Res: Serialize,
410    {
411        let outbound_tx = self.outbound_tx.clone();
412        let prev_handler = self.notification_handlers.lock().insert(
413            method,
414            Box::new(move |id, params, cx| {
415                if let Some(id) = id {
416                    if let Some(params) = serde_json::from_str(params).log_err() {
417                        let response = f(params, cx.clone());
418                        cx.foreground()
419                            .spawn({
420                                let outbound_tx = outbound_tx.clone();
421                                async move {
422                                    let response = match response.await {
423                                        Ok(result) => Response {
424                                            id,
425                                            result: Some(result),
426                                            error: None,
427                                        },
428                                        Err(error) => Response {
429                                            id,
430                                            result: None,
431                                            error: Some(Error {
432                                                message: error.to_string(),
433                                            }),
434                                        },
435                                    };
436                                    if let Some(response) = serde_json::to_vec(&response).log_err()
437                                    {
438                                        outbound_tx.try_send(response).ok();
439                                    }
440                                }
441                            })
442                            .detach();
443                    }
444                }
445            }),
446        );
447        assert!(
448            prev_handler.is_none(),
449            "registered multiple handlers for the same LSP method"
450        );
451        Subscription {
452            method,
453            notification_handlers: self.notification_handlers.clone(),
454        }
455    }
456
457    pub fn name<'a>(self: &'a Arc<Self>) -> &'a str {
458        &self.name
459    }
460
461    pub fn capabilities<'a>(self: &'a Arc<Self>) -> &'a ServerCapabilities {
462        &self.capabilities
463    }
464
465    pub fn server_id(&self) -> usize {
466        self.server_id
467    }
468
469    pub fn request<T: request::Request>(
470        &self,
471        params: T::Params,
472    ) -> impl Future<Output = Result<T::Result>>
473    where
474        T::Result: 'static + Send,
475    {
476        Self::request_internal::<T>(
477            &self.next_id,
478            &self.response_handlers,
479            &self.outbound_tx,
480            params,
481        )
482    }
483
484    fn request_internal<T: request::Request>(
485        next_id: &AtomicUsize,
486        response_handlers: &Mutex<HashMap<usize, ResponseHandler>>,
487        outbound_tx: &channel::Sender<Vec<u8>>,
488        params: T::Params,
489    ) -> impl 'static + Future<Output = Result<T::Result>>
490    where
491        T::Result: 'static + Send,
492    {
493        let id = next_id.fetch_add(1, SeqCst);
494        let message = serde_json::to_vec(&Request {
495            jsonrpc: JSON_RPC_VERSION,
496            id,
497            method: T::METHOD,
498            params,
499        })
500        .unwrap();
501
502        let send = outbound_tx
503            .try_send(message)
504            .context("failed to write to language server's stdin");
505
506        let (tx, rx) = oneshot::channel();
507        response_handlers.lock().insert(
508            id,
509            Box::new(move |result| {
510                let response = match result {
511                    Ok(response) => {
512                        serde_json::from_str(response).context("failed to deserialize response")
513                    }
514                    Err(error) => Err(anyhow!("{}", error.message)),
515                };
516                let _ = tx.send(response);
517            }),
518        );
519
520        async move {
521            send?;
522            rx.await?
523        }
524    }
525
526    pub fn notify<T: notification::Notification>(&self, params: T::Params) -> Result<()> {
527        Self::notify_internal::<T>(&self.outbound_tx, params)
528    }
529
530    fn notify_internal<T: notification::Notification>(
531        outbound_tx: &channel::Sender<Vec<u8>>,
532        params: T::Params,
533    ) -> Result<()> {
534        let message = serde_json::to_vec(&Notification {
535            jsonrpc: JSON_RPC_VERSION,
536            method: T::METHOD,
537            params,
538        })
539        .unwrap();
540        outbound_tx.try_send(message)?;
541        Ok(())
542    }
543}
544
545impl Drop for LanguageServer {
546    fn drop(&mut self) {
547        if let Some(shutdown) = self.shutdown() {
548            self.executor.spawn(shutdown).detach();
549        }
550    }
551}
552
553impl Subscription {
554    pub fn detach(mut self) {
555        self.method = "";
556    }
557}
558
559impl Drop for Subscription {
560    fn drop(&mut self) {
561        self.notification_handlers.lock().remove(self.method);
562    }
563}
564
565#[cfg(any(test, feature = "test-support"))]
566#[derive(Clone)]
567pub struct FakeLanguageServer {
568    pub server: Arc<LanguageServer>,
569    notifications_rx: channel::Receiver<(String, String)>,
570}
571
572#[cfg(any(test, feature = "test-support"))]
573impl LanguageServer {
574    pub fn full_capabilities() -> ServerCapabilities {
575        ServerCapabilities {
576            document_highlight_provider: Some(OneOf::Left(true)),
577            code_action_provider: Some(CodeActionProviderCapability::Simple(true)),
578            document_formatting_provider: Some(OneOf::Left(true)),
579            document_range_formatting_provider: Some(OneOf::Left(true)),
580            ..Default::default()
581        }
582    }
583
584    pub fn fake(cx: AsyncAppContext) -> (Self, FakeLanguageServer) {
585        Self::fake_with_capabilities(Self::full_capabilities(), cx)
586    }
587
588    pub fn fake_with_capabilities(
589        capabilities: ServerCapabilities,
590        cx: AsyncAppContext,
591    ) -> (Self, FakeLanguageServer) {
592        let (stdin_writer, stdin_reader) = async_pipe::pipe();
593        let (stdout_writer, stdout_reader) = async_pipe::pipe();
594        let (notifications_tx, notifications_rx) = channel::unbounded();
595
596        let server = Self::new_internal(
597            0,
598            stdin_writer,
599            stdout_reader,
600            Path::new("/"),
601            cx.clone(),
602            |_| {},
603        );
604        let fake = FakeLanguageServer {
605            server: Arc::new(Self::new_internal(
606                0,
607                stdout_writer,
608                stdin_reader,
609                Path::new("/"),
610                cx.clone(),
611                move |msg| {
612                    notifications_tx
613                        .try_send((msg.method.to_string(), msg.params.get().to_string()))
614                        .ok();
615                },
616            )),
617            notifications_rx,
618        };
619        fake.handle_request::<request::Initialize, _, _>({
620            let capabilities = capabilities.clone();
621            move |_, _| {
622                let capabilities = capabilities.clone();
623                async move {
624                    Ok(InitializeResult {
625                        capabilities,
626                        ..Default::default()
627                    })
628                }
629            }
630        });
631
632        (server, fake)
633    }
634}
635
636#[cfg(any(test, feature = "test-support"))]
637impl FakeLanguageServer {
638    pub fn notify<T: notification::Notification>(&self, params: T::Params) {
639        self.server.notify::<T>(params).ok();
640    }
641
642    pub async fn receive_notification<T: notification::Notification>(&mut self) -> T::Params {
643        use futures::StreamExt as _;
644
645        loop {
646            let (method, params) = self.notifications_rx.next().await.unwrap();
647            if &method == T::METHOD {
648                return serde_json::from_str::<T::Params>(&params).unwrap();
649            } else {
650                log::info!("skipping message in fake language server {:?}", params);
651            }
652        }
653    }
654
655    pub fn handle_request<T, F, Fut>(
656        &self,
657        mut handler: F,
658    ) -> futures::channel::mpsc::UnboundedReceiver<()>
659    where
660        T: 'static + request::Request,
661        T::Params: 'static + Send,
662        F: 'static + Send + FnMut(T::Params, gpui::AsyncAppContext) -> Fut,
663        Fut: 'static + Send + Future<Output = Result<T::Result>>,
664    {
665        let (responded_tx, responded_rx) = futures::channel::mpsc::unbounded();
666        self.server.remove_request_handler::<T>();
667        self.server
668            .on_request::<T, _, _>(move |params, cx| {
669                let result = handler(params, cx.clone());
670                let responded_tx = responded_tx.clone();
671                async move {
672                    cx.background().simulate_random_delay().await;
673                    let result = result.await;
674                    responded_tx.unbounded_send(()).ok();
675                    result
676                }
677            })
678            .detach();
679        responded_rx
680    }
681
682    pub fn remove_request_handler<T>(&mut self)
683    where
684        T: 'static + request::Request,
685    {
686        self.server.remove_request_handler::<T>();
687    }
688
689    pub async fn start_progress(&mut self, token: impl Into<String>) {
690        self.notify::<notification::Progress>(ProgressParams {
691            token: NumberOrString::String(token.into()),
692            value: ProgressParamsValue::WorkDone(WorkDoneProgress::Begin(Default::default())),
693        });
694    }
695
696    pub async fn end_progress(&mut self, token: impl Into<String>) {
697        self.notify::<notification::Progress>(ProgressParams {
698            token: NumberOrString::String(token.into()),
699            value: ProgressParamsValue::WorkDone(WorkDoneProgress::End(Default::default())),
700        });
701    }
702}
703
704struct ClearResponseHandlers(Arc<Mutex<HashMap<usize, ResponseHandler>>>);
705
706impl Drop for ClearResponseHandlers {
707    fn drop(&mut self) {
708        self.0.lock().clear();
709    }
710}
711
712#[cfg(test)]
713mod tests {
714    use super::*;
715    use gpui::TestAppContext;
716
717    #[ctor::ctor]
718    fn init_logger() {
719        if std::env::var("RUST_LOG").is_ok() {
720            env_logger::init();
721        }
722    }
723
724    #[gpui::test]
725    async fn test_fake(cx: &mut TestAppContext) {
726        let (server, mut fake) = LanguageServer::fake(cx.to_async());
727
728        let (message_tx, message_rx) = channel::unbounded();
729        let (diagnostics_tx, diagnostics_rx) = channel::unbounded();
730        server
731            .on_notification::<notification::ShowMessage, _>(move |params, _| {
732                message_tx.try_send(params).unwrap()
733            })
734            .detach();
735        server
736            .on_notification::<notification::PublishDiagnostics, _>(move |params, _| {
737                diagnostics_tx.try_send(params).unwrap()
738            })
739            .detach();
740
741        let server = server.initialize(None).await.unwrap();
742        server
743            .notify::<notification::DidOpenTextDocument>(DidOpenTextDocumentParams {
744                text_document: TextDocumentItem::new(
745                    Url::from_str("file://a/b").unwrap(),
746                    "rust".to_string(),
747                    0,
748                    "".to_string(),
749                ),
750            })
751            .unwrap();
752        assert_eq!(
753            fake.receive_notification::<notification::DidOpenTextDocument>()
754                .await
755                .text_document
756                .uri
757                .as_str(),
758            "file://a/b"
759        );
760
761        fake.notify::<notification::ShowMessage>(ShowMessageParams {
762            typ: MessageType::ERROR,
763            message: "ok".to_string(),
764        });
765        fake.notify::<notification::PublishDiagnostics>(PublishDiagnosticsParams {
766            uri: Url::from_str("file://b/c").unwrap(),
767            version: Some(5),
768            diagnostics: vec![],
769        });
770        assert_eq!(message_rx.recv().await.unwrap().message, "ok");
771        assert_eq!(
772            diagnostics_rx.recv().await.unwrap().uri.as_str(),
773            "file://b/c"
774        );
775
776        fake.handle_request::<request::Shutdown, _, _>(|_, _| async move { Ok(()) });
777
778        drop(server);
779        fake.receive_notification::<notification::Exit>().await;
780    }
781}