chore(tests): add more lsp tests for formatting (#14155)

1 files changed, 107 insertions, 78 deletions

diff --git a/test_util/src/lsp.rs b/test_util/src/lsp.rs
index 9d5a74eaf..c898856bf 100644
--- a/test_util/src/lsp.rs
+++ b/test_util/src/lsp.rs
@@ -4,13 +4,14 @@ use super::new_deno_dir;
 
 use anyhow::Result;
 use lazy_static::lazy_static;
+use parking_lot::Condvar;
+use parking_lot::Mutex;
 use regex::Regex;
 use serde::de;
 use serde::Deserialize;
 use serde::Serialize;
 use serde_json::json;
 use serde_json::Value;
-use std::collections::VecDeque;
 use std::io;
 use std::io::Write;
 use std::path::Path;
@@ -19,6 +20,7 @@ use std::process::ChildStdin;
 use std::process::ChildStdout;
 use std::process::Command;
 use std::process::Stdio;
+use std::sync::Arc;
 use std::time::Duration;
 use std::time::Instant;
 use tempfile::TempDir;
@@ -28,14 +30,14 @@ lazy_static! {
     Regex::new(r"(?i)^content-length:\s+(\d+)").unwrap();
 }
 
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Clone, Debug, Deserialize, Serialize)]
 pub struct LspResponseError {
   code: i32,
   message: String,
   data: Option<Value>,
 }
 
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub enum LspMessage {
   Notification(String, Option<Value>),
   Request(u64, String, Option<Value>),
@@ -64,14 +66,16 @@ impl<'a> From<&'a [u8]> for LspMessage {
   }
 }
 
-fn read_message<R>(reader: &mut R) -> Result<Vec<u8>>
+fn read_message<R>(reader: &mut R) -> Result<Option<Vec<u8>>>
 where
   R: io::Read + io::BufRead,
 {
   let mut content_length = 0_usize;
   loop {
     let mut buf = String::new();
-    reader.read_line(&mut buf)?;
+    if reader.read_line(&mut buf)? == 0 {
+      return Ok(None);
+    }
     if let Some(captures) = CONTENT_TYPE_REG.captures(&buf) {
       let content_length_match = captures
         .get(1)
@@ -85,16 +89,70 @@ where
 
   let mut msg_buf = vec![0_u8; content_length];
   reader.read_exact(&mut msg_buf)?;
-  Ok(msg_buf)
+  Ok(Some(msg_buf))
+}
+
+struct LspStdoutReader {
+  pending_messages: Arc<(Mutex<Vec<LspMessage>>, Condvar)>,
+  read_messages: Vec<LspMessage>,
+}
+
+impl LspStdoutReader {
+  pub fn new(mut buf_reader: io::BufReader<ChildStdout>) -> Self {
+    let messages: Arc<(Mutex<Vec<LspMessage>>, Condvar)> = Default::default();
+    std::thread::spawn({
+      let messages = messages.clone();
+      move || {
+        while let Ok(Some(msg_buf)) = read_message(&mut buf_reader) {
+          let msg = LspMessage::from(msg_buf.as_slice());
+          let cvar = &messages.1;
+          {
+            let mut messages = messages.0.lock();
+            messages.push(msg);
+          }
+          cvar.notify_all();
+        }
+      }
+    });
+
+    LspStdoutReader {
+      pending_messages: messages,
+      read_messages: Vec::new(),
+    }
+  }
+
+  pub fn pending_len(&self) -> usize {
+    self.pending_messages.0.lock().len()
+  }
+
+  pub fn had_message(&self, is_match: impl Fn(&LspMessage) -> bool) -> bool {
+    self.read_messages.iter().any(&is_match)
+      || self.pending_messages.0.lock().iter().any(&is_match)
+  }
+
+  pub fn read_message<R>(
+    &mut self,
+    mut get_match: impl FnMut(&LspMessage) -> Option<R>,
+  ) -> R {
+    let (msg_queue, cvar) = &*self.pending_messages;
+    let mut msg_queue = msg_queue.lock();
+    loop {
+      for i in 0..msg_queue.len() {
+        let msg = &msg_queue[i];
+        if let Some(result) = get_match(msg) {
+          let msg = msg_queue.remove(i);
+          self.read_messages.push(msg);
+          return result;
+        }
+      }
+      cvar.wait(&mut msg_queue);
+    }
+  }
 }
 
 pub struct LspClient {
   child: Child,
-  reader: io::BufReader<ChildStdout>,
-  /// Used to hold pending messages that have come out of the expected sequence
-  /// by the harness user which will be sent first when trying to consume a
-  /// message before attempting to read a new message.
-  msg_queue: VecDeque<LspMessage>,
+  reader: LspStdoutReader,
   request_id: u64,
   start: Instant,
   writer: io::BufWriter<ChildStdin>,
@@ -179,16 +237,15 @@ impl LspClient {
       command.stderr(Stdio::null());
     }
     let mut child = command.spawn()?;
-
     let stdout = child.stdout.take().unwrap();
-    let reader = io::BufReader::new(stdout);
+    let buf_reader = io::BufReader::new(stdout);
+    let reader = LspStdoutReader::new(buf_reader);
 
     let stdin = child.stdin.take().unwrap();
     let writer = io::BufWriter::new(stdin);
 
     Ok(Self {
       child,
-      msg_queue: VecDeque::new(),
       reader,
       request_id: 1,
       start: Instant::now(),
@@ -202,75 +259,47 @@ impl LspClient {
   }
 
   pub fn queue_is_empty(&self) -> bool {
-    self.msg_queue.is_empty()
+    self.reader.pending_len() == 0
   }
 
   pub fn queue_len(&self) -> usize {
-    self.msg_queue.len()
+    self.reader.pending_len()
   }
 
-  fn read(&mut self) -> Result<LspMessage> {
-    let msg_buf = read_message(&mut self.reader)?;
-    let msg = LspMessage::from(msg_buf.as_slice());
-    Ok(msg)
+  // it's flaky to assert for a notification because a notification
+  // might arrive a little later, so only provide a method for asserting
+  // that there is no notification
+  pub fn assert_no_notification(&mut self, searching_method: &str) {
+    assert!(!self.reader.had_message(|message| match message {
+      LspMessage::Notification(method, _) => method == searching_method,
+      _ => false,
+    }))
   }
 
   pub fn read_notification<R>(&mut self) -> Result<(String, Option<R>)>
   where
     R: de::DeserializeOwned,
   {
-    if !self.msg_queue.is_empty() {
-      let mut msg_queue = VecDeque::new();
-      loop {
-        match self.msg_queue.pop_front() {
-          Some(LspMessage::Notification(method, maybe_params)) => {
-            return notification_result(method, maybe_params)
-          }
-          Some(msg) => msg_queue.push_back(msg),
-          _ => break,
-        }
-      }
-      self.msg_queue = msg_queue;
-    }
-
-    loop {
-      match self.read() {
-        Ok(LspMessage::Notification(method, maybe_params)) => {
-          return notification_result(method, maybe_params)
-        }
-        Ok(msg) => self.msg_queue.push_back(msg),
-        Err(err) => return Err(err),
-      }
-    }
+    self.reader.read_message(|msg| match msg {
+      LspMessage::Notification(method, maybe_params) => Some(
+        notification_result(method.to_owned(), maybe_params.to_owned()),
+      ),
+      _ => None,
+    })
   }
 
   pub fn read_request<R>(&mut self) -> Result<(u64, String, Option<R>)>
   where
     R: de::DeserializeOwned,
   {
-    if !self.msg_queue.is_empty() {
-      let mut msg_queue = VecDeque::new();
-      loop {
-        match self.msg_queue.pop_front() {
-          Some(LspMessage::Request(id, method, maybe_params)) => {
-            return request_result(id, method, maybe_params)
-          }
-          Some(msg) => msg_queue.push_back(msg),
-          _ => break,
-        }
-      }
-      self.msg_queue = msg_queue;
-    }
-
-    loop {
-      match self.read() {
-        Ok(LspMessage::Request(id, method, maybe_params)) => {
-          return request_result(id, method, maybe_params)
-        }
-        Ok(msg) => self.msg_queue.push_back(msg),
-        Err(err) => return Err(err),
-      }
-    }
+    self.reader.read_message(|msg| match msg {
+      LspMessage::Request(id, method, maybe_params) => Some(request_result(
+        *id,
+        method.to_owned(),
+        maybe_params.to_owned(),
+      )),
+      _ => None,
+    })
   }
 
   fn write(&mut self, value: Value) -> Result<()> {
@@ -303,17 +332,17 @@ impl LspClient {
     });
     self.write(value)?;
 
-    loop {
-      match self.read() {
-        Ok(LspMessage::Response(id, maybe_result, maybe_error)) => {
-          assert_eq!(id, self.request_id);
-          self.request_id += 1;
-          return response_result(maybe_result, maybe_error);
-        }
-        Ok(msg) => self.msg_queue.push_back(msg),
-        Err(err) => return Err(err),
+    self.reader.read_message(|msg| match msg {
+      LspMessage::Response(id, maybe_result, maybe_error) => {
+        assert_eq!(*id, self.request_id);
+        self.request_id += 1;
+        Some(response_result(
+          maybe_result.to_owned(),
+          maybe_error.to_owned(),
+        ))
       }
-    }
+      _ => None,
+    })
   }
 
   pub fn write_response<V>(&mut self, id: u64, result: V) -> Result<()>
@@ -351,11 +380,11 @@ mod tests {
   fn test_read_message() {
     let msg1 = b"content-length: 11\r\n\r\nhello world";
     let mut reader1 = std::io::Cursor::new(msg1);
-    assert_eq!(read_message(&mut reader1).unwrap(), b"hello world");
+    assert_eq!(read_message(&mut reader1).unwrap().unwrap(), b"hello world");
 
     let msg2 = b"content-length: 5\r\n\r\nhello world";
     let mut reader2 = std::io::Cursor::new(msg2);
-    assert_eq!(read_message(&mut reader2).unwrap(), b"hello");
+    assert_eq!(read_message(&mut reader2).unwrap().unwrap(), b"hello");
   }
 
   #[test]