fix(inspector): allow to poll session with reentry (#16863)

This commit completely rewrites inspector session polling.

Until now, there was a single function responsible for polling inspector
sessions which could have been called when polling the "JsRuntime"
as well as from internal inspector functions. There are some cases
where it's required to have reentrant polling of sessions (eg. when 
"debugger" statement is run) which should be blocking until inspector
sends appropriate message to continue execution. This was not possible
before, because polling of sessions didn't have reentry ability.

As a consequence, session polling was split into two separate functions:
a) one to be used when polling from async context (on each tick of event
loop in "JsRuntime")
b) one to be used when polling synchronously and potentially blocking
(used by various inspector methods).

There are further cleanups and simplifications to be made in inspector
code, but this rewrite solves the problem at hand (being able to
evaluate
"debugger" JS statement and continue inspector functionality).

Co-authored-by: Bert Belder <bertbelder@gmail.com>

1 files changed, 104 insertions, 0 deletions

diff --git a/cli/tests/inspector_tests.rs b/cli/tests/inspector_tests.rs
index 407e8f071..d3041487a 100644
--- a/cli/tests/inspector_tests.rs
+++ b/cli/tests/inspector_tests.rs
@@ -1307,4 +1307,108 @@ mod inspector {
     child.kill().unwrap();
     child.wait().unwrap();
   }
+
+  // https://github.com/denoland/deno/issues/11570
+  #[tokio::test]
+  async fn inspector_repl_debugger_statement() {
+    let mut child = util::deno_cmd()
+      .arg("repl")
+      .arg(inspect_flag_with_unique_port("--inspect"))
+      .stdin(std::process::Stdio::piped())
+      .stdout(std::process::Stdio::piped())
+      .stderr(std::process::Stdio::piped())
+      .spawn()
+      .unwrap();
+
+    let stderr = child.stderr.as_mut().unwrap();
+    let mut stderr_lines = std::io::BufReader::new(stderr)
+      .lines()
+      .map(|r| r.unwrap())
+      .filter(|s| s.as_str() != "Debugger session started.");
+    let ws_url = extract_ws_url_from_stderr(&mut stderr_lines);
+
+    let (socket, response) =
+      tokio_tungstenite::connect_async(ws_url).await.unwrap();
+    assert_eq!(response.status(), 101); // Switching protocols.
+
+    let (mut socket_tx, socket_rx) = socket.split();
+    let mut socket_rx = socket_rx
+      .map(|msg| msg.unwrap().to_string())
+      .filter(|msg| {
+        let pass = !msg.starts_with(r#"{"method":"Debugger.scriptParsed","#);
+        futures::future::ready(pass)
+      })
+      .boxed_local();
+
+    let stdin = child.stdin.take().unwrap();
+
+    let stdout = child.stdout.as_mut().unwrap();
+    let mut stdout_lines = std::io::BufReader::new(stdout)
+      .lines()
+      .map(|r| r.unwrap())
+      .filter(|s| !s.starts_with("Deno "));
+
+    assert_stderr_for_inspect(&mut stderr_lines);
+    assert_eq!(
+      &stdout_lines.next().unwrap(),
+      "exit using ctrl+d, ctrl+c, or close()"
+    );
+
+    assert_inspector_messages(
+      &mut socket_tx,
+      &[
+        r#"{"id":1,"method":"Runtime.enable"}"#,
+        r#"{"id":2,"method":"Debugger.enable"}"#,
+      ],
+      &mut socket_rx,
+      &[
+        r#"{"id":1,"result":{}}"#,
+        r#"{"id":2,"result":{"debuggerId":"#,
+      ],
+      &[
+        r#"{"method":"Runtime.executionContextCreated","params":{"context":{"id":1,"#,
+      ],
+    )
+    .await;
+
+    assert_inspector_messages(
+      &mut socket_tx,
+      &[
+        r#"{"id":3,"method":"Runtime.evaluate","params":{"expression":"debugger","objectGroup":"console","includeCommandLineAPI":true,"silent":false,"contextId":1,"returnByValue":true,"generatePreview":true,"userGesture":true,"awaitPromise":false,"replMode":true}}"#,
+      ],
+      &mut socket_rx,
+      &[],
+      &[
+        r#"{"method":"Debugger.paused""#,
+      ],
+    ).await;
+    assert_inspector_messages(
+      &mut socket_tx,
+      &[
+        r#"{"id":4,"method":"Debugger.resume","params":{"terminateOnResume":false}}"#,
+      ],
+      &mut socket_rx,
+      &[
+        r#"{"id":4,"result":{}}"#,
+        r#"{"id":3,"result":{"result":{"type":"undefined"}}}"#,
+      ],
+      &[
+        r#"{"method":"Debugger.resumed""#,
+      ],
+    ).await;
+    assert_inspector_messages(
+      &mut socket_tx,
+      &[
+        r#"{"id":5,"method":"Runtime.evaluate","params":{"expression":"1","objectGroup":"console","includeCommandLineAPI":true,"silent":false,"contextId":1,"returnByValue":true,"generatePreview":true,"userGesture":true,"awaitPromise":false,"replMode":true}}"#,
+      ],
+      &mut socket_rx,
+      &[
+        r#"{"id":5,"result":{"result":{"type":"number","value":1,"description":"1"}}}"#,
+      ],
+      &[
+      ],
+    ).await;
+    drop(stdin);
+    child.wait().unwrap();
+  }
 }