diff --git a/crates/editor/src/editor_tests.rs b/crates/editor/src/editor_tests.rs index 93a8a33f6e8..f04c9234379 100644 --- a/crates/editor/src/editor_tests.rs +++ b/crates/editor/src/editor_tests.rs @@ -22175,28 +22175,28 @@ async fn test_strip_whitespace_and_format_via_lsp(cx: &mut TestAppContext) { cx.lsp .set_request_handler::({ move |_, _| { + assert_eq!( + buffer_changes.lock().get(1..), + Some( + [ + ( + lsp::Range::new(lsp::Position::new(0, 3), lsp::Position::new(0, 4)), + String::new(), + ), + ( + lsp::Range::new(lsp::Position::new(2, 5), lsp::Position::new(2, 6)), + String::new(), + ), + ( + lsp::Range::new(lsp::Position::new(3, 4), lsp::Position::new(3, 4)), + language::LineEnding::default().as_str().to_owned(), + ), + ] + .as_slice(), + ), + ); // Insert blank lines between each line of the buffer. async move { - // TODO: this assertion is not reliably true. Currently nothing guarantees that we deliver - // DidChangedTextDocument to the LSP before sending the formatting request. - // assert_eq!( - // &buffer_changes.lock()[1..], - // &[ - // ( - // lsp::Range::new(lsp::Position::new(0, 3), lsp::Position::new(0, 4)), - // "".into() - // ), - // ( - // lsp::Range::new(lsp::Position::new(2, 5), lsp::Position::new(2, 6)), - // "".into() - // ), - // ( - // lsp::Range::new(lsp::Position::new(3, 4), lsp::Position::new(3, 4)), - // "\n".into() - // ), - // ] - // ); - Ok(Some(vec![ lsp::TextEdit { range: lsp::Range::new( diff --git a/crates/editor/src/inlays/inlay_hints.rs b/crates/editor/src/inlays/inlay_hints.rs index cc9c02eee6e..3afd8c12adf 100644 --- a/crates/editor/src/inlays/inlay_hints.rs +++ b/crates/editor/src/inlays/inlay_hints.rs @@ -7292,13 +7292,18 @@ let c = 3;"# }) .await .unwrap(); - let editor = - cx.add_window(|window, cx| Editor::for_buffer(buffer, Some(project), window, cx)); + let mut fake_server_a = fake_servers_a.next().await.unwrap(); + let mut fake_server_b = fake_servers_b.next().await.unwrap(); + fake_server_a + .receive_notification::() + .await; + fake_server_b + .receive_notification::() + .await; cx.executor().run_until_parked(); - let _fake_server_a = fake_servers_a.next().await.unwrap(); - let fake_server_b = fake_servers_b.next().await.unwrap(); - + let editor = + cx.add_window(|window, cx| Editor::for_buffer(buffer, Some(project), window, cx)); editor .update(cx, |editor, window, cx| { editor.set_visible_line_count(50.0, window, cx); diff --git a/crates/lsp/src/lsp.rs b/crates/lsp/src/lsp.rs index 2b602d2cd05..ce68fa25eac 100644 --- a/crates/lsp/src/lsp.rs +++ b/crates/lsp/src/lsp.rs @@ -109,14 +109,16 @@ pub struct LanguageServerBinaryOptions { pub pre_release: bool, } -struct NotificationSerializer(Box String + Send + Sync>); +enum OutboundMessage { + Serialized(String), + PendingSerialization(Box String + Send + Sync>), +} /// A running language server process. pub struct LanguageServer { server_id: LanguageServerId, next_id: AtomicI32, - outbound_tx: async_channel::Sender, - notification_tx: async_channel::Sender, + outbound_tx: async_channel::Sender, name: LanguageServerName, version: Option, process_name: Arc, @@ -600,7 +602,8 @@ impl LanguageServer { Stderr: AsyncRead + Unpin + Send + 'static, F: Fn(&NotificationOrRequest) -> bool + 'static + Send + Sync + Clone, { - let (outbound_tx, outbound_rx) = async_channel::unbounded::(); + let (outbound_tx, outbound_rx) = async_channel::unbounded::(); + let (serialized_tx, serialized_rx) = async_channel::unbounded::(); let (output_done_tx, output_done_rx) = barrier::channel(); let notification_handlers = Arc::new(Mutex::new(HashMap::<_, NotificationHandler>::default())); @@ -622,7 +625,10 @@ impl LanguageServer { result: None, }; if let Ok(response) = serde_json::to_string(&response) { - response_channel.send(response).await.ok(); + response_channel + .send(OutboundMessage::Serialized(response)) + .await + .ok(); } } } @@ -663,7 +669,7 @@ impl LanguageServer { let output_task = cx.background_spawn({ Self::handle_outgoing_messages( stdin, - outbound_rx, + serialized_rx, output_done_tx, response_handlers.clone(), io_handlers.clone(), @@ -676,25 +682,22 @@ impl LanguageServer { } .into(); - let (notification_tx, notification_rx) = - async_channel::unbounded::(); - cx.background_spawn({ - let outbound_tx = outbound_tx.clone(); - async move { - while let Ok(serializer) = notification_rx.recv().await { - let serialized = (serializer.0)(); - let Ok(_) = outbound_tx.send(serialized).await else { - return; - }; - } - outbound_tx.close(); + cx.background_spawn(async move { + while let Ok(message) = outbound_rx.recv().await { + let serialized = match message { + OutboundMessage::Serialized(message) => message, + OutboundMessage::PendingSerialization(serialize) => serialize(), + }; + let Ok(_) = serialized_tx.send(serialized).await else { + return; + }; } + serialized_tx.close(); }) .detach(); Self { server_id, notification_handlers, - notification_tx, response_handlers, pending_respond_tasks, io_handlers, @@ -1203,13 +1206,11 @@ impl LanguageServer { let next_id = AtomicI32::new(self.next_id.load(SeqCst)); let outbound_tx = self.outbound_tx.clone(); let executor = self.executor.clone(); - let notification_serializers = self.notification_tx.clone(); let mut output_done = self.output_done_rx.lock().take().unwrap(); let shutdown_request = Self::request_internal::( &next_id, &response_handlers, &outbound_tx, - ¬ification_serializers, &executor, SERVER_SHUTDOWN_TIMEOUT, (), @@ -1246,8 +1247,8 @@ impl LanguageServer { }; response_handlers.lock().take(); - Self::notify_internal::(¬ification_serializers, ()).ok(); - notification_serializers.close(); + Self::notify_internal::(&outbound_tx, ()).ok(); + outbound_tx.close(); if !shutdown_timed_out { select! { _ = output_done.recv().fuse() => {}, @@ -1391,7 +1392,9 @@ impl LanguageServer { if let Some(response) = serde_json::to_string(&response).log_err() { - outbound_tx.try_send(response).ok(); + outbound_tx + .try_send(OutboundMessage::Serialized(response)) + .ok(); } pending_respond_tasks.lock().remove(&id); } @@ -1411,7 +1414,9 @@ impl LanguageServer { )), }; if let Some(response) = serde_json::to_string(&response).log_err() { - outbound_tx.try_send(response).ok(); + outbound_tx + .try_send(OutboundMessage::Serialized(response)) + .ok(); } } } @@ -1512,7 +1517,6 @@ impl LanguageServer { &self.next_id, &self.response_handlers, &self.outbound_tx, - &self.notification_tx, &self.executor, request_timeout, params, @@ -1522,8 +1526,7 @@ impl LanguageServer { fn request_internal_with_timer( next_id: &AtomicI32, response_handlers: &Arc>>>, - outbound_tx: &async_channel::Sender, - notification_serializers: &async_channel::Sender, + outbound_tx: &async_channel::Sender, executor: &BackgroundExecutor, timer: U, params: T::Params, @@ -1571,11 +1574,11 @@ impl LanguageServer { }); let send = outbound_tx - .try_send(message) + .try_send(OutboundMessage::Serialized(message)) .context("failed to write to language server's stdin"); let response_handlers = Arc::clone(response_handlers); - let notification_serializers = notification_serializers.downgrade(); + let outbound_tx = outbound_tx.downgrade(); let started = Instant::now(); LspRequest::new(id, async move { if let Err(e) = handle_response { @@ -1586,9 +1589,9 @@ impl LanguageServer { } let cancel_on_drop = gpui_util::defer(move || { - if let Some(notification_serializers) = notification_serializers.upgrade() { + if let Some(outbound_tx) = outbound_tx.upgrade() { Self::notify_internal::( - ¬ification_serializers, + &outbound_tx, CancelParams { id: NumberOrString::Number(id), }, @@ -1632,8 +1635,7 @@ impl LanguageServer { fn request_internal( next_id: &AtomicI32, response_handlers: &Arc>>>, - outbound_tx: &async_channel::Sender, - notification_serializers: &async_channel::Sender, + outbound_tx: &async_channel::Sender, executor: &BackgroundExecutor, request_timeout: Duration, params: T::Params, @@ -1646,7 +1648,6 @@ impl LanguageServer { next_id, response_handlers, outbound_tx, - notification_serializers, executor, Self::request_timeout_future(executor.clone(), request_timeout), params, @@ -1679,15 +1680,14 @@ impl LanguageServer { /// /// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#notificationMessage) pub fn notify(&self, params: T::Params) -> Result<()> { - let outbound = self.notification_tx.clone(); - Self::notify_internal::(&outbound, params) + Self::notify_internal::(&self.outbound_tx, params) } fn notify_internal( - outbound_tx: &async_channel::Sender, + outbound_tx: &async_channel::Sender, params: T::Params, ) -> Result<()> { - let serializer = NotificationSerializer(Box::new(move || { + let message = OutboundMessage::PendingSerialization(Box::new(move || { serde_json::to_string(&Notification { jsonrpc: JSON_RPC_VERSION, method: T::METHOD, @@ -1696,7 +1696,7 @@ impl LanguageServer { .unwrap() })); - outbound_tx.send_blocking(serializer)?; + outbound_tx.send_blocking(message)?; Ok(()) } @@ -2260,6 +2260,69 @@ mod tests { fake.receive_notification::().await; } + #[gpui::test] + async fn test_notifications_and_requests_preserve_send_order(cx: &mut TestAppContext) { + let (server, mut stdout, output) = shutdown_test_server(None, cx); + let uri = Uri::from_str("file:///test.py").expect("valid document URI"); + let notify_change = |version, text: &str| { + server + .notify::(DidChangeTextDocumentParams { + text_document: VersionedTextDocumentIdentifier::new(uri.clone(), version), + content_changes: vec![TextDocumentContentChangeEvent { + range: None, + range_length: None, + text: text.to_owned(), + }], + }) + .expect("document change must be queued"); + }; + + notify_change(1, "α = 1\n"); + let formatting = server.request::( + DocumentFormattingParams { + text_document: TextDocumentIdentifier::new(uri.clone()), + options: FormattingOptions { + tab_size: 4, + insert_spaces: true, + ..FormattingOptions::default() + }, + work_done_progress_params: WorkDoneProgressParams::default(), + }, + DEFAULT_LSP_REQUEST_TIMEOUT, + ); + notify_change(2, "α = 2\n"); + cx.run_until_parked(); + + let expected = [ + r#"{"jsonrpc":"2.0","method":"textDocument/didChange","params":{"textDocument":{"uri":"file:///test.py","version":1},"contentChanges":[{"text":"α = 1\n"}]}}"#, + r#"{"jsonrpc":"2.0","id":0,"method":"textDocument/formatting","params":{"textDocument":{"uri":"file:///test.py"},"options":{"tabSize":4,"insertSpaces":true}}}"#, + r#"{"jsonrpc":"2.0","method":"textDocument/didChange","params":{"textDocument":{"uri":"file:///test.py","version":2},"contentChanges":[{"text":"α = 2\n"}]}}"#, + ] + .into_iter() + .flat_map(framed_test_message) + .collect::>(); + assert_eq!( + String::from_utf8(output.lock().bytes.clone()).expect("UTF-8 output"), + String::from_utf8(expected.clone()).expect("UTF-8 expected output"), + ); + + stdout + .write_all(&framed_test_message( + r#"{"jsonrpc":"2.0","id":0,"result":[]}"#, + )) + .await + .expect("formatting response must reach stdout"); + assert_eq!( + formatting + .await + .into_response() + .expect("formatting response"), + Some(Vec::new()), + ); + cx.run_until_parked(); + assert_eq!(output.lock().bytes, expected); + } + #[gpui::test] async fn test_shutdown_bounds_output_drain(cx: &mut TestAppContext) { for remaining_flushes in [Some(0), Some(1), None] {