feat(jupyter): add handling for comms

cli/js/40_jupyter.js

@@ -402,19 +402,42 @@ function enableJupyter() {
   const { op_jupyter_broadcast, op_jupyter_comm_recv, op_jupyter_comm_open } =
     core.ops;
 
-  function commOpen(commId, targetName) {
+  function comm(commId, targetName, msgCallback) {
     op_jupyter_comm_open(commId, targetName);
-  }
 
-  async function commRecv(commId) {
-    const [data, buffers] = await op_jupyter_comm_recv(commId);
+    let closed = false;
+
+    // TODO(bartlomieju): return something, so we can close this comm.
+    (async () => {
+      while (true) {
+        const [data, buffers] = await op_jupyter_comm_recv(commId);
+
+        if (!data) {
+          closed = true;
+          break;
+        }
+
+        msgCallback?.({
+          ...data,
+          buffers,
+        });
+      }
+    })();
 
-    if (!data) {
-      return undefined;
-    }
     return {
-      ...data,
-      buffers,
+      close() {
+        if (closed) {
+          return;
+        }
+
+        closed = true;
+      },
+      send(data) {
+        return broadcast("comm_msg", {
+          comm_id: commId,
+          data: data,
+        });
+      },
     };
   }
 
@@ -493,8 +516,7 @@ function enableJupyter() {
 
   globalThis.Deno.jupyter = {
     broadcast,
-    commRecv,
-    commOpen,
+    comm,
     display,
     format,
     md,

cli/tools/jupyter/server.rs

@@ -217,6 +217,7 @@ impl JupyterServer {
       .send_iopub(messaging::Status::busy().as_child_of(parent))
       .await?;
 
+    eprintln!("msg type on shell {}", msg.message_type());
     match msg.content {
       JupyterMessageContent::ExecuteRequest(execute_request) => {
         self
@@ -375,6 +376,7 @@ impl JupyterServer {
         // NOTE: This will belong on the stdin channel, not the shell channel
       }
       JupyterMessageContent::CommOpen(comm) => {
+        eprintln!("comm_open");
         self
           .comm_container
           .create(&comm.comm_id.0, &comm.target_name, None);
@@ -390,6 +392,7 @@ impl JupyterServer {
         //   .await?;
       }
       JupyterMessageContent::CommInfoRequest(_req) => {
+        eprintln!("comm_open");
         connection
           .send(
             messaging::CommInfoReply {