feat(ext/node): better "node:inspector" support

Cargo.toml

@@ -395,3 +395,6 @@ opt-level = 3
 opt-level = 3
 [profile.release.package.zstd-sys]
 opt-level = 3
+
+[patch.crates-io]
+deno_core = { path = "../deno_core/core" }

bar.js

@@ -0,0 +1 @@
+setInterval(() => console.log("hello"), 5_000);

cli/tools/repl/session.rs

@@ -41,6 +41,7 @@ use deno_core::serde_json::Value;
 use deno_core::unsync::spawn;
 use deno_core::url::Url;
 use deno_core::LocalInspectorSession;
+use deno_core::LocalInspectorSessionOptions;
 use deno_core::PollEventLoopOptions;
 use deno_graph::source::ResolutionMode;
 use deno_graph::source::Resolver;
@@ -204,7 +205,10 @@ impl ReplSession {
     test_event_receiver: TestEventReceiver,
   ) -> Result<Self, AnyError> {
     let language_server = ReplLanguageServer::new_initialized().await?;
-    let mut session = worker.create_inspector_session();
+    let mut session =
+      worker.create_inspector_session(LocalInspectorSessionOptions {
+        kind: deno_core::InspectorSessionKind::LocalBlocking,
+      });
 
     worker
       .js_runtime

cli/worker.rs

@@ -72,14 +72,14 @@ pub trait ModuleLoaderFactory: Send + Sync {
 }
 
 #[async_trait::async_trait(?Send)]
-pub trait HmrRunner: Send + Sync {
+pub trait HmrRunner {
   async fn start(&mut self) -> Result<(), AnyError>;
   async fn stop(&mut self) -> Result<(), AnyError>;
   async fn run(&mut self) -> Result<(), AnyError>;
 }
 
 #[async_trait::async_trait(?Send)]
-pub trait CoverageCollector: Send + Sync {
+pub trait CoverageCollector {
   async fn start_collecting(&mut self) -> Result<(), AnyError>;
   async fn stop_collecting(&mut self) -> Result<(), AnyError>;
 }
@@ -368,7 +368,11 @@ impl CliMainWorker {
       return Ok(None);
     };
 
-    let session = self.worker.create_inspector_session();
+    let session = self.worker.create_inspector_session(
+      deno_core::LocalInspectorSessionOptions {
+        kind: deno_core::InspectorSessionKind::LocalBlocking,
+      },
+    );
 
     let mut hmr_runner = setup_hmr_runner(session);
 
@@ -392,7 +396,11 @@ impl CliMainWorker {
       return Ok(None);
     };
 
-    let session = self.worker.create_inspector_session();
+    let session = self.worker.create_inspector_session(
+      deno_core::LocalInspectorSessionOptions {
+        kind: deno_core::InspectorSessionKind::LocalBlocking,
+      },
+    );
     let mut coverage_collector = create_coverage_collector(session);
     self
       .worker

ext/node/lib.rs

@@ -370,6 +370,9 @@ deno_core::extension!(deno_node,
     ops::ipc::op_node_ipc_unref,
     ops::process::op_node_process_kill,
     ops::process::op_process_abort,
+    ops::inspector::op_inspector_disconnect,
+    ops::inspector::op_inspector_post,
+    ops::inspector::op_inspector_get_message_from_v8,
   ],
   esm_entry_point = "ext:deno_node/02_init.js",
   esm = [

ext/node/ops/inspector.rs

@@ -0,0 +1,29 @@
+use deno_core::error::AnyError;
+use deno_core::op2;
+use deno_core::serde_json;
+use deno_core::LocalInspectorSessionRaw;
+
+#[op2(fast)]
+pub fn op_inspector_disconnect(#[cppgc] session: &LocalInspectorSessionRaw) {
+  session.disconnect();
+}
+
+#[op2]
+pub fn op_inspector_post(
+  #[cppgc] session: &LocalInspectorSessionRaw,
+  #[smi] id: i32,
+  #[string] method: String,
+  #[serde] params: Option<serde_json::Value>,
+) -> Result<(), AnyError> {
+  session.post_message(id, &method, params);
+  Ok(())
+}
+
+#[op2(async)]
+#[string]
+pub async fn op_inspector_get_message_from_v8(
+  #[cppgc] session: &LocalInspectorSessionRaw,
+) -> Option<String> {
+  let maybe_inspector_message = session.receive_from_v8_session().await;
+  maybe_inspector_message.map(|msg| msg.content)
+}

ext/node/ops/mod.rs

@@ -7,6 +7,7 @@ pub mod fs;
 pub mod http;
 pub mod http2;
 pub mod idna;
+pub mod inspector;
 pub mod ipc;
 pub mod os;
 pub mod process;

ext/node/polyfills/inspector.ts

@@ -1,45 +1,145 @@
 // Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
 // Copyright Joyent and Node contributors. All rights reserved. MIT license.
 
+import {
+  op_create_inspector_session,
+  op_inspector_disconnect,
+  op_inspector_get_message_from_v8,
+  op_inspector_post,
+} from "ext:core/ops";
+import {
+  ERR_INSPECTOR_ALREADY_CONNECTED,
+  ERR_INSPECTOR_CLOSED,
+  ERR_INSPECTOR_COMMAND,
+  ERR_INSPECTOR_NOT_CONNECTED,
+} from "ext:deno_node/internal/errors.ts";
+import {
+  validateFunction,
+  validateObject,
+  validateString,
+} from "ext:deno_node/internal/validators.mjs";
+import { nextTick } from "ext:deno_node/_next_tick.ts";
 import { EventEmitter } from "node:events";
+import { emitWarning } from "node:process";
 import { notImplemented } from "ext:deno_node/_utils.ts";
-import { primordials } from "ext:core/mod.js";
+import { core, primordials } from "ext:core/mod.js";
 
 const {
   SafeMap,
+  JSONParse,
 } = primordials;
 
 class Session extends EventEmitter {
-  #connection = null;
+  // TODO(bartlomieju): how do we type CppGc objects?
+  #connection: any = null;
   #nextId = 1;
   #messageCallbacks = new SafeMap();
 
   /** Connects the session to the inspector back-end. */
-  connect() {
-    notImplemented("inspector.Session.prototype.connect");
+  connect(): void {
+    if (this.#connection) {
+      throw new ERR_INSPECTOR_ALREADY_CONNECTED("The inspector session");
+    }
+
+    this.#connection = op_create_inspector_session();
+
+    // Start listening for messages - this is using "unrefed" op
+    // so that listening for notifications doesn't block the event loop.
+    // When posting a message another promise should be started that is refed.
+    (async () => {
+      while (true) {
+        await this.#listenForMessage(true);
+      }
+    })();
+  }
+
+  async #listenForMessage(unref: boolean) {
+    let message: string;
+    try {
+      const opPromise = op_inspector_get_message_from_v8(this.#connection);
+      if (unref) {
+        core.unrefOpPromise(opPromise);
+      }
+      message = await opPromise;
+      this.#onMessage(message);
+    } catch (e) {
+      emitWarning(e);
+    }
+  }
+
+  #onMessage(message: string) {
+    core.print("received message" + message + "\n", true);
+    const parsed = JSONParse(message);
+    if (parsed.id) {
+      const callback = this.#messageCallbacks.get(parsed.id);
+      this.#messageCallbacks.delete(parsed.id);
+      if (callback) {
+        if (parsed.error) {
+          return callback(
+            new ERR_INSPECTOR_COMMAND(parsed.error.code, parsed.error.message),
+          );
+        }
+
+        callback(null, parsed.result);
+      }
+    } else {
+      this.emit(parsed.method, parsed);
+      this.emit("inspectorNotification", parsed);
+    }
   }
 
   /** Connects the session to the main thread
    * inspector back-end. */
-  connectToMainThread() {
+  connectToMainThread(): void {
     notImplemented("inspector.Session.prototype.connectToMainThread");
   }
 
   /** Posts a message to the inspector back-end. */
   post(
-    _method: string,
-    _params?: Record<string, unknown>,
-    _callback?: (...args: unknown[]) => void,
-  ) {
-    notImplemented("inspector.Session.prototype.post");
+    method: string,
+    params: Record<string, unknown> | null,
+    callback?: (...args: unknown[]) => void,
+  ): void {
+    validateString(method, "method");
+    if (!callback && typeof params === "function") {
+      callback = params;
+      params = null;
+    }
+    if (params) {
+      validateObject(params, "params");
+    }
+    if (callback) {
+      validateFunction(callback, "callback");
+    }
+
+    if (!this.#connection) {
+      throw new ERR_INSPECTOR_NOT_CONNECTED();
+    }
+    const id = this.#nextId++;
+    if (callback) {
+      this.#messageCallbacks.set(id, callback);
+    }
+    // TODO(bartlomieju): Should this ignore errors? Check with Node impl
+    op_inspector_post(this.#connection, id, method, params);
+    this.#listenForMessage(false);
   }
 
   /** Immediately closes the session, all pending
    * message callbacks will be called with an
    * error.
    */
   disconnect() {
-    notImplemented("inspector.Session.prototype.disconnect");
+    if (!this.#connection) {
+      return;
+    }
+    op_inspector_disconnect(this.#connection);
+    this.#connection = null;
+    const remainingCallbacks = this.#messageCallbacks.values();
+    for (const callback of remainingCallbacks) {
+      nextTick(callback, new ERR_INSPECTOR_CLOSED());
+    }
+    this.#messageCallbacks.clear();
+    this.#nextId = 1;
   }
 }
 

foo.mjs

@@ -0,0 +1,52 @@
+import inspector from "node:inspector";
+
+let resolve_;
+const deferred = new Promise((resolve) => {
+  resolve_ = resolve;
+});
+
+const session = new inspector.Session();
+session.connect();
+console.log("Connected");
+
+session.on("inspectorNotification", (ev) => {
+  console.log("Received notification", ev.method);
+});
+
+session.post("Runtime.enable", undefined, (err, params) => {
+  console.log("Runtime.enable, err:", err, "params:", params);
+});
+
+session.post(
+  "Runtime.evaluate",
+  {
+    // contextId: 1,
+    expression: "new Promise(resolve => setTimeout(resolve, 1000))",
+    awaitPromise: true,
+  },
+  (err, params) => {
+    console.log("Runtime.evaluate, err:", err, "params:", params);
+    // resolve_();
+  },
+);
+
+// await deferred;
+session.disconnect();
+session.disconnect();
+
+session.connect();
+// console.log("Connected again");
+// session.post("Runtime.enable", undefined, (err, params) => {
+//   console.log("Runtime.enable, err:", err, "params:", params);
+// });
+
+// session.post(
+//   "Runtime.evaluate",
+//   {
+//     // contextId: 1,
+//     expression: "1 + 2",
+//   },
+//   (err, params) => {
+//     console.log("Runtime.evaluate, err:", err, "params:", params);
+//   },
+// );