diff --git a/crates/ide-jsonrpc/src/lib.rs b/crates/ide-jsonrpc/src/lib.rs index 638fd75f6..caa1cde24 100644 --- a/crates/ide-jsonrpc/src/lib.rs +++ b/crates/ide-jsonrpc/src/lib.rs @@ -498,20 +498,30 @@ impl Peer { }); } - // Notifications are delivered in order on a dedicated thread. - let (notif_tx, notif_rx) = std::sync::mpsc::channel::<(String, Value)>(); + // Notifications are delivered in order on a dedicated thread. Requests start from the same + // queue, so a request handler sees every notification that arrived before it. + let (notif_tx, notif_rx) = std::sync::mpsc::channel::(); { let handler = handler.clone(); + let inner = inner.clone(); + let runtime = tokio::runtime::Handle::current(); std::thread::Builder::new() .name("jsonrpc-notify".into()) .spawn(move || { - while let Ok((method, params)) = notif_rx.recv() { - let h = handler.clone(); - let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - h.on_notification(&method, params) - })); - if result.is_err() { - tracing::error!("jsonrpc notification handler panicked on {method}"); + while let Ok(msg) = notif_rx.recv() { + match msg { + Incoming::Notification(method, params) => { + let h = handler.clone(); + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + h.on_notification(&method, params) + })); + if result.is_err() { + tracing::error!("jsonrpc notification handler panicked on {method}"); + } + } + Incoming::Request(id, method, params) => { + runtime.spawn(handle_request(inner.clone(), handler.clone(), id, method, params)); + } } } }) @@ -543,7 +553,7 @@ impl Peer { match msg { Ok(Some(body)) => { seen_message.store(true, Ordering::Relaxed); - dispatch(&inner, &handler, ¬if_tx, &body); + dispatch(&inner, ¬if_tx, &body); } Ok(None) => break, Err(e) => { @@ -680,12 +690,28 @@ fn parse_id(id: &Value) -> Option { } } -fn dispatch( - inner: &Arc, - handler: &Arc, - notif_tx: &std::sync::mpsc::Sender<(String, Value)>, - body: &[u8], -) { +enum Incoming { + Notification(String, Value), + Request(Value, String, Value), +} + +async fn handle_request(inner: Arc, handler: Arc, id: Value, method: String, params: Value) { + let m = method.clone(); + let result = tokio::task::spawn_blocking(move || handler.on_request(&m, params)).await; + let reply = match result { + Ok(Ok(result)) => json!({ "jsonrpc": "2.0", "id": id, "result": result }), + Ok(Err((code, message))) => { + json!({ "jsonrpc": "2.0", "id": id, "error": { "code": code, "message": message } }) + } + Err(_) => json!({ + "jsonrpc": "2.0", "id": id, + "error": { "code": codes::INTERNAL_ERROR, "message": format!("handler for {method} panicked") } + }), + }; + let _ = inner.send_value(&reply); +} + +fn dispatch(inner: &Arc, notif_tx: &std::sync::mpsc::Sender, body: &[u8]) { tracing::trace!(target: "ide_jsonrpc::wire", "<-- {}", String::from_utf8_lossy(body)); let msg: Value = match serde_json::from_slice(body) { Ok(v) => v, @@ -704,26 +730,10 @@ fn dispatch( let params = obj.remove("params").unwrap_or(Value::Null); match id { Some(id) if !id.is_null() => { - let handler = handler.clone(); - let inner = inner.clone(); - tokio::spawn(async move { - let m = method.clone(); - let result = tokio::task::spawn_blocking(move || handler.on_request(&m, params)).await; - let reply = match result { - Ok(Ok(result)) => json!({ "jsonrpc": "2.0", "id": id, "result": result }), - Ok(Err((code, message))) => { - json!({ "jsonrpc": "2.0", "id": id, "error": { "code": code, "message": message } }) - } - Err(_) => json!({ - "jsonrpc": "2.0", "id": id, - "error": { "code": codes::INTERNAL_ERROR, "message": format!("handler for {method} panicked") } - }), - }; - let _ = inner.send_value(&reply); - }); + let _ = notif_tx.send(Incoming::Request(id, method, params)); } _ => { - let _ = notif_tx.send((method, params)); + let _ = notif_tx.send(Incoming::Notification(method, params)); } } }