refactor(core): cleanup Inspector implementation (#12962)

1 files changed, 16 insertions, 15 deletions

diff --git a/runtime/inspector_server.rs b/runtime/inspector_server.rs
index 99aff0644..04a8db959 100644
--- a/runtime/inspector_server.rs
+++ b/runtime/inspector_server.rs
@@ -16,9 +16,11 @@ use deno_core::futures::task::Poll;
 use deno_core::serde_json;
 use deno_core::serde_json::json;
 use deno_core::serde_json::Value;
+use deno_core::InspectorMsg;
 use deno_core::InspectorSessionProxy;
 use deno_core::JsRuntime;
 use deno_websocket::tokio_tungstenite::tungstenite;
+use deno_websocket::tokio_tungstenite::WebSocketStream;
 use std::cell::RefCell;
 use std::collections::HashMap;
 use std::convert::Infallible;
@@ -169,13 +171,12 @@ fn handle_ws_request(
       eprintln!("Inspector server failed to upgrade to WS connection");
       return;
     };
-    let websocket =
-      deno_websocket::tokio_tungstenite::WebSocketStream::from_raw_socket(
-        upgraded,
-        tungstenite::protocol::Role::Server,
-        None,
-      )
-      .await;
+    let websocket = WebSocketStream::from_raw_socket(
+      upgraded,
+      tungstenite::protocol::Role::Server,
+      None,
+    )
+    .await;
 
     // The 'outbound' channel carries messages sent to the websocket.
     let (outbound_tx, outbound_rx) = mpsc::unbounded();
@@ -330,26 +331,26 @@ async fn server(
 /// 'futures' crate, therefore they can't participate in Tokio's cooperative
 /// task yielding.
 async fn pump_websocket_messages(
-  websocket: deno_websocket::tokio_tungstenite::WebSocketStream<
-    hyper::upgrade::Upgraded,
-  >,
-  inbound_tx: UnboundedSender<Vec<u8>>,
-  outbound_rx: UnboundedReceiver<(Option<i32>, String)>,
+  websocket: WebSocketStream<hyper::upgrade::Upgraded>,
+  inbound_tx: UnboundedSender<String>,
+  outbound_rx: UnboundedReceiver<InspectorMsg>,
 ) {
   let (websocket_tx, websocket_rx) = websocket.split();
 
   let outbound_pump = outbound_rx
-    .map(|(_maybe_call_id, msg)| tungstenite::Message::text(msg))
+    .map(|msg| tungstenite::Message::text(msg.content))
     .map(Ok)
     .forward(websocket_tx)
     .map_err(|_| ());
 
   let inbound_pump = async move {
     let _result = websocket_rx
-      .map_ok(|msg| msg.into_data())
       .map_err(AnyError::from)
       .map_ok(|msg| {
-        let _ = inbound_tx.unbounded_send(msg);
+        // Messages that cannot be converted to strings are ignored.
+        if let Ok(msg_text) = msg.into_text() {
+          let _ = inbound_tx.unbounded_send(msg_text);
+        }
       })
       .try_collect::<()>()
       .await;