1 files changed, 61 insertions, 29 deletions

diff --git a/cli/tests/integration/inspector_tests.rs b/cli/tests/integration/inspector_tests.rs
index 067963786..35ff01403 100644
--- a/cli/tests/integration/inspector_tests.rs
+++ b/cli/tests/integration/inspector_tests.rs
@@ -1,15 +1,16 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
 
+use deno_core::anyhow::anyhow;
 use deno_core::error::AnyError;
-use deno_core::futures::prelude::*;
-use deno_core::futures::stream::SplitSink;
-use deno_core::futures::stream::SplitStream;
 use deno_core::serde_json;
 use deno_core::serde_json::json;
 use deno_core::url;
 use deno_runtime::deno_fetch::reqwest;
 use deno_runtime::deno_websocket::tokio_tungstenite;
-use deno_runtime::deno_websocket::tokio_tungstenite::tungstenite;
+use fastwebsockets::FragmentCollector;
+use fastwebsockets::Frame;
+use hyper::upgrade::Upgraded;
+use hyper::Request;
 use std::io::BufRead;
 use test_util as util;
 use test_util::TempDir;
@@ -17,18 +18,20 @@ use tokio::net::TcpStream;
 use util::http_server;
 use util::DenoChild;
 
+struct SpawnExecutor;
+
+impl<Fut> hyper::rt::Executor<Fut> for SpawnExecutor
+where
+  Fut: std::future::Future + Send + 'static,
+  Fut::Output: Send + 'static,
+{
+  fn execute(&self, fut: Fut) {
+    tokio::task::spawn(fut);
+  }
+}
+
 struct InspectorTester {
-  socket_tx: SplitSink<
-    tokio_tungstenite::WebSocketStream<
-      tokio_tungstenite::MaybeTlsStream<TcpStream>,
-    >,
-    tungstenite::Message,
-  >,
-  socket_rx: SplitStream<
-    tokio_tungstenite::WebSocketStream<
-      tokio_tungstenite::MaybeTlsStream<TcpStream>,
-    >,
-  >,
+  socket: FragmentCollector<Upgraded>,
   notification_filter: Box<dyn FnMut(&str) -> bool + 'static>,
   child: DenoChild,
   stderr_lines: Box<dyn Iterator<Item = String>>,
@@ -52,17 +55,42 @@ impl InspectorTester {
     let mut stderr_lines =
       std::io::BufReader::new(stderr).lines().map(|r| r.unwrap());
 
-    let ws_url = extract_ws_url_from_stderr(&mut stderr_lines);
+    let uri = extract_ws_url_from_stderr(&mut stderr_lines);
+
+    let domain = &uri.host().unwrap().to_string();
+    let port = &uri.port().unwrap_or(match uri.scheme() {
+      "wss" | "https" => 443,
+      _ => 80,
+    });
+    let addr = format!("{domain}:{port}");
+
+    let stream = TcpStream::connect(addr).await.unwrap();
+
+    let host = uri.host_str().unwrap();
+
+    let req = Request::builder()
+      .method("GET")
+      .uri(uri.path())
+      .header("Host", host)
+      .header(hyper::header::UPGRADE, "websocket")
+      .header(hyper::header::CONNECTION, "Upgrade")
+      .header(
+        "Sec-WebSocket-Key",
+        fastwebsockets::handshake::generate_key(),
+      )
+      .header("Sec-WebSocket-Version", "13")
+      .body(hyper::Body::empty())
+      .unwrap();
 
     let (socket, response) =
-      tokio_tungstenite::connect_async(ws_url).await.unwrap();
-    assert_eq!(response.status(), 101); // Switching protocols.
+      fastwebsockets::handshake::client(&SpawnExecutor, req, stream)
+        .await
+        .unwrap();
 
-    let (socket_tx, socket_rx) = socket.split();
+    assert_eq!(response.status(), 101); // Switching protocols.
 
     Self {
-      socket_tx,
-      socket_rx,
+      socket: FragmentCollector::new(socket),
       notification_filter: Box::new(notification_filter),
       child,
       stderr_lines: Box::new(stderr_lines),
@@ -74,10 +102,10 @@ impl InspectorTester {
     // TODO(bartlomieju): add graceful error handling
     for msg in messages {
       let result = self
-        .socket_tx
-        .send(msg.to_string().into())
+        .socket
+        .write_frame(Frame::text(msg.to_string().into_bytes()))
         .await
-        .map_err(|e| e.into());
+        .map_err(|e| anyhow!(e));
       self.handle_error(result);
     }
   }
@@ -111,8 +139,9 @@ impl InspectorTester {
 
   async fn recv(&mut self) -> String {
     loop {
-      let result = self.socket_rx.next().await.unwrap().map_err(|e| e.into());
-      let message = self.handle_error(result).to_string();
+      let result = self.socket.read_frame().await.map_err(|e| anyhow!(e));
+      let message =
+        String::from_utf8(self.handle_error(result).payload).unwrap();
       if (self.notification_filter)(&message) {
         return message;
       }
@@ -236,7 +265,7 @@ fn skip_check_line(
     let mut line = stderr_lines.next().unwrap();
     line = util::strip_ansi_codes(&line).to_string();
 
-    if line.starts_with("Check") {
+    if line.starts_with("Check") || line.starts_with("Download") {
       continue;
     }
 
@@ -514,8 +543,11 @@ async fn inspector_does_not_hang() {
   }
 
   // Check that we can gracefully close the websocket connection.
-  tester.socket_tx.close().await.unwrap();
-  tester.socket_rx.for_each(|_| async {}).await;
+  tester
+    .socket
+    .write_frame(Frame::close_raw(vec![]))
+    .await
+    .unwrap();
 
   assert_eq!(&tester.stdout_lines.next().unwrap(), "done");
   assert!(tester.child.wait().unwrap().success());