refactor(core): allow to listen for notifications in LocalInspectorSession (#17040)

cli/tools/repl/session.rs

@@ -12,7 +12,9 @@ use deno_ast::DiagnosticsError;
 use deno_ast::ImportsNotUsedAsValues;
 use deno_ast::ModuleSpecifier;
 use deno_core::error::AnyError;
+use deno_core::futures::channel::mpsc::UnboundedReceiver;
 use deno_core::futures::FutureExt;
+use deno_core::futures::StreamExt;
 use deno_core::serde_json;
 use deno_core::serde_json::Value;
 use deno_core::LocalInspectorSession;
@@ -92,6 +94,10 @@ pub struct ReplSession {
   pub language_server: ReplLanguageServer,
   has_initialized_node_runtime: bool,
   referrer: ModuleSpecifier,
+  // FIXME(bartlomieju): this field should be used to listen
+  // for "exceptionThrown" notifications
+  #[allow(dead_code)]
+  notification_rx: UnboundedReceiver<Value>,
 }
 
 impl ReplSession {
@@ -113,8 +119,11 @@ impl ReplSession {
     // Enabling the runtime domain will always send trigger one executionContextCreated for each
     // context the inspector knows about so we grab the execution context from that since
     // our inspector does not support a default context (0 is an invalid context id).
-    let mut context_id: u64 = 0;
-    for notification in session.notifications() {
+    let context_id: u64;
+    let mut notification_rx = session.take_notification_rx();
+
+    loop {
+      let notification = notification_rx.next().await.unwrap();
       let method = notification.get("method").unwrap().as_str().unwrap();
       let params = notification.get("params").unwrap();
       if method == "Runtime.executionContextCreated" {
@@ -127,6 +136,7 @@ impl ReplSession {
           .as_bool()
           .unwrap());
         context_id = context.get("id").unwrap().as_u64().unwrap();
+        break;
       }
     }
     assert_ne!(context_id, 0);
@@ -141,6 +151,7 @@ impl ReplSession {
       language_server,
       has_initialized_node_runtime: false,
       referrer,
+      notification_rx,
     };
 
     // inject prelude

core/inspector.rs

@@ -696,7 +696,8 @@ pub struct LocalInspectorSession {
   v8_session_rx: UnboundedReceiver<InspectorMsg>,
   response_tx_map: HashMap<i32, oneshot::Sender<serde_json::Value>>,
   next_message_id: i32,
-  notification_queue: Vec<Value>,
+  notification_tx: UnboundedSender<Value>,
+  notification_rx: Option<UnboundedReceiver<Value>>,
 }
 
 impl LocalInspectorSession {
@@ -707,19 +708,20 @@ impl LocalInspectorSession {
     let response_tx_map = HashMap::new();
     let next_message_id = 0;
 
-    let notification_queue = Vec::new();
+    let (notification_tx, notification_rx) = mpsc::unbounded::<Value>();
 
     Self {
       v8_session_tx,
       v8_session_rx,
       response_tx_map,
       next_message_id,
-      notification_queue,
+      notification_tx,
+      notification_rx: Some(notification_rx),
     }
   }
 
-  pub fn notifications(&mut self) -> Vec<Value> {
-    self.notification_queue.split_off(0)
+  pub fn take_notification_rx(&mut self) -> UnboundedReceiver<Value> {
+    self.notification_rx.take().unwrap()
   }
 
   pub async fn post_message<T: serde::Serialize>(
@@ -795,7 +797,8 @@ impl LocalInspectorSession {
         .unwrap();
     } else {
       let message = serde_json::from_str(&inspector_msg.content).unwrap();
-      self.notification_queue.push(message);
+      // Ignore if the receiver has been dropped.
+      let _ = self.notification_tx.unbounded_send(message);
     }
   }
 }