adding awareness event when open and close websockets (#246)

Co-authored-by: Jialin Zhang <[email protected]>

Author: jzhang20133

jupyter_collaboration/app.py

@@ -11,7 +11,7 @@
 from .handlers import DocSessionHandler, YDocWebSocketHandler
 from .loaders import FileLoaderMapping
 from .stores import SQLiteYStore
-from .utils import EVENTS_SCHEMA_PATH
+from .utils import AWARENESS_EVENTS_SCHEMA_PATH, EVENTS_SCHEMA_PATH
 from .websocketserver import JupyterWebsocketServer
 
 
@@ -60,6 +60,7 @@ class YDocExtension(ExtensionApp):
     def initialize(self):
         super().initialize()
         self.serverapp.event_logger.register_event_schema(EVENTS_SCHEMA_PATH)
+        self.serverapp.event_logger.register_event_schema(AWARENESS_EVENTS_SCHEMA_PATH)
 
     def initialize_settings(self):
         self.settings.update(

jupyter_collaboration/events/awareness.yaml

@@ -0,0 +1,33 @@
+"$id": https://schema.jupyter.org/jupyter_collaboration/awareness/v1
+"$schema": "http://json-schema.org/draft-07/schema"
+version: 1
+title: Collaborative awareness events
+personal-data: true
+description: |
+  Awareness events emitted from server-side during a collaborative session.
+type: object
+required:
+  - roomid
+  - username
+  - action
+properties:
+  roomid:
+    type: string
+    description: |
+      Room ID. Usually composed by the file type, format and ID.
+  username:
+    type: string
+    description: |
+      The name of the user who joined or left room.
+  action:
+    enum:
+      - join
+      - leave
+    description: |
+      Possible values:
+      1. join
+      2. leave
+  msg:
+    type: string
+    description: |
+      Optional event message.

jupyter_collaboration/handlers.py

@@ -21,6 +21,7 @@
 from .loaders import FileLoaderMapping
 from .rooms import DocumentRoom, TransientRoom
 from .utils import (
+    JUPYTER_COLLABORATION_AWARENESS_EVENTS_URI,
     JUPYTER_COLLABORATION_EVENTS_URI,
     LogLevel,
     MessageType,
@@ -184,6 +185,7 @@ async def open(self, room_id):
             try:
                 # Initialize the room
                 await self.room.initialize()
+                self._emit_awareness_event(self.current_user.username, "join")
             except Exception as e:
                 _, _, file_id = decode_file_path(self._room_id)
                 file = self._file_loaders[file_id]
@@ -205,6 +207,9 @@ async def open(self, room_id):
                     await self._clean_room()
 
             self._emit(LogLevel.INFO, "initialize", "New client connected.")
+        else:
+            if self.room.room_id != "JupyterLab:globalAwareness":
+                self._emit_awareness_event(self.current_user.username, "join")
 
     async def send(self, message):
         """
@@ -284,6 +289,8 @@ def on_close(self) -> None:
             # keep the document for a while in case someone reconnects
             self.log.info("Cleaning room: %s", self._room_id)
             self.room.cleaner = asyncio.create_task(self._clean_room())
+        if self.room.room_id != "JupyterLab:globalAwareness":
+            self._emit_awareness_event(self.current_user.username, "leave")
 
     def _emit(self, level: LogLevel, action: str | None = None, msg: str | None = None) -> None:
         _, _, file_id = decode_file_path(self._room_id)
@@ -297,6 +304,13 @@ def _emit(self, level: LogLevel, action: str | None = None, msg: str | None = No
 
         self.event_logger.emit(schema_id=JUPYTER_COLLABORATION_EVENTS_URI, data=data)
 
+    def _emit_awareness_event(self, username: str, action: str, msg: str | None = None) -> None:
+        data = {"roomid": self._room_id, "username": username, "action": action}
+        if msg:
+            data["msg"] = msg
+
+        self.event_logger.emit(schema_id=JUPYTER_COLLABORATION_AWARENESS_EVENTS_URI, data=data)
+
     async def _clean_room(self) -> None:
         """
         Async task for cleaning up the resources.

jupyter_collaboration/utils.py

@@ -1,12 +1,17 @@
 # Copyright (c) Jupyter Development Team.
 # Distributed under the terms of the Modified BSD License.
 
-import pathlib
 from enum import Enum, IntEnum
+from pathlib import Path
 from typing import Tuple
 
+EVENTS_FOLDER_PATH = Path(__file__).parent / "events"
 JUPYTER_COLLABORATION_EVENTS_URI = "https://schema.jupyter.org/jupyter_collaboration/session/v1"
-EVENTS_SCHEMA_PATH = pathlib.Path(__file__).parent / "events" / "session.yaml"
+EVENTS_SCHEMA_PATH = EVENTS_FOLDER_PATH / "session.yaml"
+JUPYTER_COLLABORATION_AWARENESS_EVENTS_URI = (
+    "https://schema.jupyter.org/jupyter_collaboration/awareness/v1"
+)
+AWARENESS_EVENTS_SCHEMA_PATH = EVENTS_FOLDER_PATH / "awareness.yaml"
 
 
 class MessageType(IntEnum):

tests/test_handlers.py

@@ -7,6 +7,7 @@
 from asyncio import Event, sleep
 from typing import Any
 
+from jupyter_events.logger import EventLogger
 from jupyter_ydoc import YUnicode
 from pycrdt_websocket import WebsocketProvider
 
@@ -83,3 +84,50 @@ def _on_document_change(target: str, e: Any) -> None:
         await sleep(0.1)
 
     assert doc.source == content
+
+
+async def test_room_handler_doc_client_should_emit_awareness_event(
+    rtc_create_file, rtc_connect_doc_client, jp_serverapp
+):
+    path, content = await rtc_create_file("test.txt", "test")
+
+    event = Event()
+
+    def _on_document_change(target: str, e: Any) -> None:
+        if target == "source":
+            event.set()
+
+    doc = YUnicode()
+    doc.observe(_on_document_change)
+
+    listener_was_called = False
+    collected_data = []
+
+    async def my_listener(logger: EventLogger, schema_id: str, data: dict) -> None:
+        nonlocal listener_was_called
+        collected_data.append(data)
+        listener_was_called = True
+
+    event_logger = jp_serverapp.event_logger
+    event_logger.add_listener(
+        schema_id="https://schema.jupyter.org/jupyter_collaboration/awareness/v1",
+        listener=my_listener,
+    )
+
+    async with await rtc_connect_doc_client("text", "file", path) as ws, WebsocketProvider(
+        doc.ydoc, ws
+    ):
+        await event.wait()
+        await sleep(0.1)
+
+    fim = jp_serverapp.web_app.settings["file_id_manager"]
+
+    assert doc.source == content
+    assert listener_was_called is True
+    assert len(collected_data) == 2
+    assert collected_data[0]["action"] == "join"
+    assert collected_data[0]["roomid"] == "text:file:" + fim.get_id("test.txt")
+    assert collected_data[0]["username"] is not None
+    assert collected_data[1]["action"] == "leave"
+    assert collected_data[1]["roomid"] == "text:file:" + fim.get_id("test.txt")
+    assert collected_data[1]["username"] is not None