Fixing bugs.

Author: ellisonbg

jupyter_rtc_core/app.py

@@ -23,11 +23,9 @@ class RtcExtensionApp(ExtensionApp):
         (r"api/collaboration/room/(.*)", YRoomWebsocket),
         # # handler that just adds compatibility with Jupyter Collaboration's frontend
         # (r"api/collaboration/session/(.*)", YRoomSessionHandler),
-        (r"api/fileid/index", FileIDIndexHandler)
+        (r"api/fileid/index", FileIDIndexHandler),
+        *outputs_handlers
     ]
-    
-    for handler in outputs_handlers:
-        handlers.append(handler)
 
     yroom_manager_class = Type(
         klass=YRoomManager,

jupyter_rtc_core/outputs/manager.py

@@ -29,7 +29,6 @@ def _default_outputs_path(self):
 
     def _ensure_path(self, file_id, cell_id):
         nested_dir = self.outputs_path / file_id / cell_id
-        self.log.info(f"Creating directory: {nested_dir}")
         nested_dir.mkdir(parents=True, exist_ok=True)
 
     def _build_path(self, file_id, cell_id=None, output_index=None):
@@ -42,7 +41,6 @@ def _build_path(self, file_id, cell_id=None, output_index=None):
 
     def get_output(self, file_id, cell_id, output_index):
         """Get an outputs by file_id, cell_id, and output_index."""
-        self.log.info(f"OutputsManager.get: {file_id} {cell_id} {output_index}")
         path = self._build_path(file_id, cell_id, output_index)
         if not os.path.isfile(path):
             raise FileNotFoundError(f"The output file doesn't exist: {path}")
@@ -60,12 +58,15 @@ def get_stream(self, file_id, cell_id):
         return output
 
     def write(self, file_id, cell_id, output):
-        """Write a new output for file_id and cell_id."""
-        self.log.info(f"OutputsManager.write: {file_id} {cell_id} {output}")
-        result = self.write_output(file_id, cell_id, output)
+        """Write a new output for file_id and cell_id.
+        
+        Returns a placeholder output (pycrdt.Map) or None if no placeholder
+        output should be written to the ydoc.
+        """
+        placeholder = self.write_output(file_id, cell_id, output)
         if output["output_type"] == "stream" and self.stream_limit is not None:
-            result = self.write_stream(file_id, cell_id, output)
-        return result
+            placeholder = self.write_stream(file_id, cell_id, output, placeholder)
+        return placeholder
 
     def write_output(self, file_id, cell_id, output):
         self._ensure_path(file_id, cell_id)
@@ -77,9 +78,10 @@ def write_output(self, file_id, cell_id, output):
         with open(path, "w", encoding="utf-8") as f:
             f.write(data)
         url = f"/api/outputs/{file_id}/{cell_id}/{index}.output"
-        return Map({})
+        self.log.info(f"Wrote output: {url}")
+        return create_placeholder_output(output["output_type"], url)
 
-    def write_stream(self, file_id, cell_id, output) -> Map:
+    def write_stream(self, file_id, cell_id, output, placeholder) -> Map:
         # How many stream outputs have been written for this cell previously
         count = self._stream_count.get(cell_id, 0)
 
@@ -91,15 +93,16 @@ def write_stream(self, file_id, cell_id, output) -> Map:
         with open(path, "a", encoding="utf-8") as f:
             f.write(text)
         url = f"/api/outputs/{file_id}/{cell_id}/stream"
+        self.log.info(f"Wrote stream: {url}")
 
         # Increment the count
         count = count + 1
         self._stream_count[cell_id] = count
 
         # Now create the placeholder output
         if count < self.stream_limit:
-            # Return the original if we haven't reached the limit
-            placeholder = Map({})
+            # Return the original placeholder if we haven't reached the limit
+            placeholder = placeholder
         elif count == self.stream_limit:
             # Return a link to the full stream output
             placeholder = Map({
@@ -109,6 +112,7 @@ def write_stream(self, file_id, cell_id, output) -> Map:
                 }
             })
         elif count > self.stream_limit:
+            # Return None to indicate that no placeholder should be written to the ydoc
             placeholder = None
         return placeholder
 
@@ -123,4 +127,33 @@ def clear(self, file_id, cell_id=None):
             except KeyError:
                 pass
             path = self._build_path(file_id, cell_id)
-        shutil.rmtree(path)
+        try:
+            shutil.rmtree(path)
+        except FileNotFoundError:
+            pass
+
+
+def create_placeholder_output(output_type: str, url: str):
+    metadata = dict(url=url)
+    if output_type == "stream":
+        output = Map({
+            "output_type": "stream",
+            "text": "",
+            "metadata": metadata
+        })
+    elif output_type == "display_data":
+        output = Map({
+            "output_type": "display_data",
+            "metadata": metadata
+        })
+    elif output_type == "execute_result":
+        output = Map({
+            "output_type": "execute_result",
+            "metadata": metadata
+        })
+    elif output_type == "error":
+        output = Map({
+            "output_type": "error",
+            "metadata": metadata
+        })
+    return output

jupyter_rtc_core/outputs/output_processor.py

@@ -22,7 +22,7 @@ class OutputProcessor(LoggingConfigurable):
     def settings(self):
         """A shortcut for the Tornado web app settings."""
         #      self.KernelClient.KernelManager.AsyncMultiKernelManager.ServerApp
-        return self.parent.parent.parent.web_app.settings
+        return self.parent.parent.parent.parent.web_app.settings
 
     @property
     def kernel_client(self):
@@ -45,9 +45,9 @@ def file_id_manager(self):
         return self.settings["file_id_manager"]
 
     @property
-    def jupyter_server_ydoc(self):
+    def yroom_manager(self):
         """A shortcut for the jupyter server ydoc manager."""
-        return self.settings["jupyter_server_ydoc"]
+        return self.settings["yroom_manager"]
 
     def clear(self, cell_id=None):
         """Clear the state of the output processor.
@@ -126,17 +126,12 @@ def process_outgoing_message(self, dmsg: dict):
         The content has not been deserialized yet as we need to verify we
         should process it.
         """
-        self.log.info("Process outgoing message")
         msg_type = dmsg["header"]["msg_type"]
-        self.log.info(f"msg_type: {msg_type}")
         if msg_type not in ("stream", "display_data", "execute_result", "error"):
             return dmsg
         msg_id = dmsg["parent_header"]["msg_id"]
-        self.log.info(f"msg_type: {msg_type}")
         content = self.parent.session.unpack(dmsg["content"])
-        self.log.info(f"content: {content}")
         cell_id = self.get_cell_id(msg_id)
-        self.log.info(f"cell_id: {cell_id}")
         if cell_id is None:
             # This is valid as cell_id is optional
             return dmsg
@@ -145,9 +140,11 @@ def process_outgoing_message(self, dmsg: dict):
 
     async def output_task(self, msg_type, cell_id, content):
         """A coroutine to handle output messages."""
-        self.log.info(f"output_task: {msg_type} {cell_id}")
         try:
-            kernel_session = await self.session_manager.get_session(kernel_id=self.kernel_id)
+            # TODO: The session manager may have multiple notebooks connected to the kernel
+            # but currently get_session only returns the first. We need to fix this and
+            # find the notebook with the right cell_id.
+            kernel_session = await self.session_manager.get_session(kernel_id=self.parent.parent.kernel_id)
         except Exception as e:
             self.log.error(f"An exception occurred when processing output for cell {cell_id}")
             self.log.exception(e)
@@ -156,37 +153,33 @@ async def output_task(self, msg_type, cell_id, content):
             path = kernel_session["path"]
 
         file_id = self.file_id_manager.get_id(path)
-        self.log.info(f"Output for file_id, cell_id: {file_id} {cell_id}")
         if file_id is None:
             self.log.error(f"Could not find file_id for path: {path}")
             return
         self._file_id = file_id
-        try:
-            notebook = await self.jupyter_server_ydoc.get_document(
-                path=path,
-                copy=False,
-                file_format='json',
-                content_type='notebook'
-            )
-        except Exception as e:
-            self.log.error(f"An exception occurred when processing output for cell {cell_id}")
-            self.log.exception(e)
-            return
-        cells = notebook.ycells
-
-        cell_index, target_cell = self.find_cell(cell_id, cells)
-        if target_cell is None:
-            # This is valid as cell_id is optional
-            return
 
         # Convert from the message spec to the nbformat output structure
         if self.use_outputs_service:
             output = self.transform_output(msg_type, content, ydoc=False)
             output = self.outputs_manager.write(file_id, cell_id, output)
         else:
             output = self.transform_output(msg_type, content, ydoc=True)
-        if output is not None:
+
+        # Get the notebook ydoc from the room
+        room_id = f"json:notebook:{file_id}"
+        room = self.yroom_manager.get_room(room_id)
+        if room is None:
+            self.log.error(f"YRoom not found: {room_id}")
+            return
+        notebook = await room.get_jupyter_ydoc()
+        self.log.info(f"Notebook: {notebook}")
+ 
+        # Write the outputs to the ydoc cell.
+        cells = notebook.ycells
+        cell_index, target_cell = self.find_cell(cell_id, cells)
+        if target_cell is not None and output is not None:
             target_cell["outputs"].append(output)
+            self.log.info(f"Write output to ydoc: {path} {cell_id} {output}")
 
     def find_cell(self, cell_id, cells):
         """Find a cell with a given cell_id in the list of cells.