feat(unstable): add `Deno.jupyter.broadcast` API (#20656)

Closes https://github.com/denoland/deno/issues/20591

---------

Co-authored-by: Kyle Kelley <rgbkrk@gmail.com>

1 files changed, 11 insertions, 45 deletions

diff --git a/cli/tools/jupyter/mod.rs b/cli/tools/jupyter/mod.rs
index 7d7104060..fb0860e36 100644
--- a/cli/tools/jupyter/mod.rs
+++ b/cli/tools/jupyter/mod.rs
@@ -2,24 +2,23 @@
 
 use crate::args::Flags;
 use crate::args::JupyterFlags;
+use crate::ops;
 use crate::tools::repl;
 use crate::util::logger;
 use crate::CliFactory;
 use deno_core::anyhow::Context;
 use deno_core::error::AnyError;
-use deno_core::futures::channel::mpsc;
-use deno_core::op2;
+use deno_core::located_script_name;
 use deno_core::resolve_url_or_path;
 use deno_core::serde::Deserialize;
 use deno_core::serde_json;
-use deno_core::Op;
-use deno_core::OpState;
 use deno_runtime::permissions::Permissions;
 use deno_runtime::permissions::PermissionsContainer;
+use tokio::sync::mpsc;
 
 mod install;
-mod jupyter_msg;
-mod server;
+pub(crate) mod jupyter_msg;
+pub(crate) mod server;
 
 pub async fn kernel(
   flags: Flags,
@@ -59,7 +58,7 @@ pub async fn kernel(
   let npm_resolver = factory.npm_resolver().await?.clone();
   let resolver = factory.resolver().await?.clone();
   let worker_factory = factory.create_cli_main_worker_factory().await?;
-  let (stdio_tx, stdio_rx) = mpsc::unbounded();
+  let (stdio_tx, stdio_rx) = mpsc::unbounded_channel();
 
   let conn_file =
     std::fs::read_to_string(&connection_filepath).with_context(|| {
@@ -77,11 +76,15 @@ pub async fn kernel(
     .create_custom_worker(
       main_module.clone(),
       permissions,
-      vec![deno_jupyter::init_ops(stdio_tx)],
+      vec![ops::jupyter::deno_jupyter::init_ops(stdio_tx)],
       Default::default(),
     )
     .await?;
   worker.setup_repl().await?;
+  worker.execute_script_static(
+    located_script_name!(),
+    "Deno[Deno.internal].enableJupyter();",
+  )?;
   let worker = worker.into_main_worker();
   let repl_session =
     repl::ReplSession::initialize(cli_options, npm_resolver, resolver, worker)
@@ -92,43 +95,6 @@ pub async fn kernel(
   Ok(())
 }
 
-deno_core::extension!(deno_jupyter,
-  options = {
-    sender: mpsc::UnboundedSender<server::StdioMsg>,
-  },
-  middleware = |op| match op.name {
-    "op_print" => op_print::DECL,
-    _ => op,
-  },
-  state = |state, options| {
-    state.put(options.sender);
-  },
-);
-
-#[op2(fast)]
-pub fn op_print(
-  state: &mut OpState,
-  #[string] msg: &str,
-  is_err: bool,
-) -> Result<(), AnyError> {
-  let sender = state.borrow_mut::<mpsc::UnboundedSender<server::StdioMsg>>();
-
-  if is_err {
-    if let Err(err) =
-      sender.unbounded_send(server::StdioMsg::Stderr(msg.into()))
-    {
-      eprintln!("Failed to send stderr message: {}", err);
-    }
-    return Ok(());
-  }
-
-  if let Err(err) = sender.unbounded_send(server::StdioMsg::Stdout(msg.into()))
-  {
-    eprintln!("Failed to send stdout message: {}", err);
-  }
-  Ok(())
-}
-
 #[derive(Debug, Deserialize)]
 pub struct ConnectionSpec {
   ip: String,