@@ -1,7 +1,7 @@
11use crate::{Error, Result};
22use anyhow::Context;
33use jsonrpc_core as jsonrpc;
4−use log::{debug, error, info, warn};
4+use log::{error, info};
55use serde::{Deserialize, Serialize};
66use serde_json::Value;
77use std::collections::HashMap;
@@ -11,7 +11,7 @@ use tokio::{
1111 process::{ChildStderr, ChildStdin, ChildStdout},
1212 sync::{
1313 mpsc::{unbounded_channel, Sender, UnboundedReceiver, UnboundedSender},
14− Mutex,
14+ Mutex, Notify,
1515 },
1616};
1717
@@ -51,9 +51,11 @@ impl Transport {
5151 ) -> (
5252 UnboundedReceiver<(usize, jsonrpc::Call)>,
5353 UnboundedSender<Payload>,
54+ Arc<Notify>,
5455 ) {
5556 let (client_tx, rx) = unbounded_channel();
5657 let (tx, client_rx) = unbounded_channel();
58+ let notify = Arc::new(Notify::new());
5759
5860 let transport = Self {
5961 id,
@@ -64,9 +66,14 @@ impl Transport {
6466
6567 tokio::spawn(Self::recv(transport.clone(), server_stdout, client_tx));
6668 tokio::spawn(Self::err(transport.clone(), server_stderr));
67− tokio::spawn(Self::send(transport, server_stdin, client_rx));
68−
69− (rx, tx)
69+ tokio::spawn(Self::send(
70+ transport,
71+ server_stdin,
72+ client_rx,
73+ notify.clone(),
74+ ));
75+
76+ (rx, tx, notify)
7077 }
7178
7279 async fn recv_server_message(
@@ -82,7 +89,8 @@ impl Transport {
8289
8390 // debug!("<- header {:?}", buffer);
8491
85− if header.is_empty() {
92+ if buffer == "\r\n" {
93+ // look for an empty CRLF line
8694 break;
8795 }
8896
@@ -99,7 +107,8 @@ impl Transport {
99107 // Workaround: Some non-conformant language servers will output logging and other garbage
100108 // into the same stream as JSON-RPC messages. This can also happen from shell scripts that spawn
101109 // the server. Skip such lines and log a warning.
102− warn!("Failed to parse header: {:?}", header);
110+
111+ // warn!("Failed to parse header: {:?}", header);
103112 }
104113 }
105114 }
@@ -261,15 +270,67 @@ impl Transport {
261270 transport: Arc<Self>,
262271 mut server_stdin: BufWriter<ChildStdin>,
263272 mut client_rx: UnboundedReceiver<Payload>,
273+ initialize_notify: Arc<Notify>,
264274 ) {
265− while let Some(msg) = client_rx.recv().await {
266− match transport
267− .send_payload_to_server(&mut server_stdin, msg)
268− .await
269− {
270− Ok(_) => {}
271− Err(err) => {
272− error!("err: <- {:?}", err);
275+ let mut pending_messages: Vec<Payload> = Vec::new();
276+ let mut is_pending = true;
277+
278+ // Determine if a message is allowed to be sent early
279+ fn is_initialize(payload: &Payload) -> bool {
280+ use lsp_types::{
281+ notification::{Initialized, Notification},
282+ request::{Initialize, Request},
283+ };
284+ match payload {
285+ Payload::Request {
286+ value: jsonrpc::MethodCall { method, .. },
287+ ..
288+ } if method == Initialize::METHOD => true,
289+ Payload::Notification(jsonrpc::Notification { method, .. })
290+ if method == Initialized::METHOD =>
291+ {
292+ true
293+ }
294+ _ => false,
295+ }
296+ }
297+
298+ // TODO: events that use capabilities need to do the right thing
299+
300+ loop {
301+ tokio::select! {
302+ biased;
303+ _ = initialize_notify.notified() => { // TODO: notified is technically not cancellation safe
304+ // server successfully initialized
305+ is_pending = false;
306+ // drain the pending queue and send payloads to server
307+ for msg in pending_messages.drain(..) {
308+ log::info!("Draining pending message {:?}", msg);
309+ match transport.send_payload_to_server(&mut server_stdin, msg).await {
310+ Ok(_) => {}
311+ Err(err) => {
312+ error!("err: <- {:?}", err);
313+ }
314+ }
315+ }
316+ }
317+ msg = client_rx.recv() => {
318+ if let Some(msg) = msg {
319+ if is_pending && !is_initialize(&msg) {
320+ log::info!("Language server not initialized, delaying request");
321+ pending_messages.push(msg);
322+ } else {
323+ match transport.send_payload_to_server(&mut server_stdin, msg).await {
324+ Ok(_) => {}
325+ Err(err) => {
326+ error!("err: <- {:?}", err);
327+ }
328+ }
329+ }
330+ } else {
331+ // channel closed
332+ break;
333+ }
273334 }
274335 }
275336 }