Creates a dialog to answer messages

jupyter_collaboration/app.py

@@ -33,7 +33,7 @@ class YDocExtension(ExtensionApp):
     )
 
     document_cleanup_delay = Float(
-        60,
+        10,
         allow_none=True,
         config=True,
         help="""The delay in seconds to keep a document in memory in the back-end after all clients

jupyter_collaboration/handlers.py

@@ -24,7 +24,6 @@
     JUPYTER_COLLABORATION_EVENTS_URI,
     LogLevel,
     MessageType,
-    RoomMessages,
     decode_file_path,
 )
 from .websocketserver import JupyterWebsocketServer
@@ -320,7 +319,7 @@ async def _clean_room(self) -> None:
         file = self._file_loaders[file_id]
         if file.number_of_subscriptions == 0:
             self.log.info("Deleting file %s", file.path)
-            del self._file_loaders[file_id]
+            await self._file_loaders.remove(file_id)
             self._emit(LogLevel.INFO, "clean", "Loader deleted.")
 
     def check_origin(self, origin):

jupyter_collaboration/loaders.py

@@ -167,7 +167,8 @@ async def maybe_save_content(self, model: dict[str, Any]) -> dict[str, Any]:
             )
 
             if model["last_modified"] == m["last_modified"]:
-                return await self.save_content(model)
+                self._log.info("Saving file: %s", path)
+                return await ensure_async(self._contents_manager.save(model, path))
 
             else:
                 # file changed on disk, raise an error

jupyter_collaboration/rooms.py

@@ -53,6 +53,7 @@ def __init__(
         self._save_delay = save_delay
 
         self._update_lock = asyncio.Lock()
+        self._outofband_lock = asyncio.Lock()
         self._initialization_lock = asyncio.Lock()
         self._cleaner: asyncio.Task | None = None
         self._saving_document: asyncio.Task | None = None
@@ -160,7 +161,7 @@ async def initialize(self) -> None:
 
     async def handle_msg(self, data: bytes) -> None:
         msg_type = data[0]
-        msg_id = data[1:].decode()
+        msg_id = data[2:].decode()
 
         # Use a lock to prevent handling responses from multiple clients
         # at the same time
@@ -171,9 +172,9 @@ async def handle_msg(self, data: bytes) -> None:
 
             try:
                 ans = None
-                if msg_type == RoomMessages.RESTORE:
+                if msg_type == RoomMessages.RELOAD:
                     # Restore the room with the content from disk
-                    await self._restore()
+                    await self._load_document()
                     ans = RoomMessages.DOC_OVERWRITTEN
 
                 elif msg_type == RoomMessages.OVERWRITE:
@@ -185,7 +186,8 @@ async def handle_msg(self, data: bytes) -> None:
                     # Remove the lock and broadcast the resolution
                     self._messages.pop(msg_id)
                     data = msg_id.encode()
-                    self._broadcast_msg(
+                    self._outofband_lock.release()
+                    await self._broadcast_msg(
                         bytes([MessageType.ROOM, ans]) + write_var_uint(len(data)) + data
                     )
 
@@ -230,6 +232,9 @@ async def _on_content_change(self, event: str, args: dict[str, Any]) -> None:
                 event (str): Type of change.
                 args (dict): A dictionary with format, type, last_modified.
         """
+        if self._outofband_lock.locked():
+            return
+
         if event == "metadata" and (
             self._last_modified is None or self._last_modified < args["last_modified"]
         ):
@@ -238,8 +243,9 @@ async def _on_content_change(self, event: str, args: dict[str, Any]) -> None:
 
             msg_id = str(uuid.uuid4())
             self._messages[msg_id] = asyncio.Lock()
+            await self._outofband_lock.acquire()
             data = msg_id.encode()
-            self._broadcast_msg(
+            await self._broadcast_msg(
                 bytes([MessageType.ROOM, RoomMessages.FILE_CHANGED])
                 + write_var_uint(len(data))
                 + data
@@ -270,7 +276,7 @@ def _on_document_change(self, target: str, event: Any) -> None:
 
         self._saving_document = asyncio.create_task(self._maybe_save_document())
 
-    async def _restore(self) -> None:
+    async def _load_document(self) -> None:
         try:
             model = await self._file.load_content(self._file_format, self._file_type, True)
         except Exception as e:
@@ -362,9 +368,9 @@ async def _maybe_save_document(self) -> None:
             self.log.error(msg, exc_info=e)
             self._emit(LogLevel.ERROR, None, msg)
 
-    def _broadcast_msg(self, msg) -> None:
+    async def _broadcast_msg(self, msg: bytes) -> None:
         for client in self.clients:
-            client.send(msg)
+            await client.send(msg)
 
 
 class TransientRoom(YRoom):

jupyter_collaboration/utils.py

@@ -17,7 +17,7 @@ class MessageType(IntEnum):
 
 
 class RoomMessages(IntEnum):
-    RESTORE = 0
+    RELOAD = 0
     OVERWRITE = 1
     FILE_CHANGED = 2
     FILE_OVERWRITTEN = 3

packages/docprovider/src/awareness.ts

@@ -14,12 +14,9 @@ import * as decoding from 'lib0/decoding';
 import * as encoding from 'lib0/encoding';
 import { WebsocketProvider } from 'y-websocket';
 
+import { MessageType } from './utils';
 import { IAwarenessProvider } from './tokens';
 
-export enum MessageType {
-  CHAT = 125
-}
-
 export interface IContent {
   type: string;
   body: string;

packages/docprovider/src/utils.ts

@@ -0,0 +1,17 @@
+/* -----------------------------------------------------------------------------
+| Copyright (c) Jupyter Development Team.
+| Distributed under the terms of the Modified BSD License.
+|----------------------------------------------------------------------------*/
+
+export enum MessageType {
+  ROOM = 124,
+  CHAT = 125
+}
+
+export enum RoomMessage {
+  RELOAD = 0,
+  OVERWRITE = 1,
+  FILE_CHANGED = 2,
+  FILE_OVERWRITTEN = 3,
+  DOC_OVERWRITTEN = 4
+}

packages/docprovider/src/yprovider.ts

@@ -13,10 +13,13 @@ import { Signal } from '@lumino/signaling';
 
 import { DocumentChange, YDocument } from '@jupyter/ydoc';
 
+import * as decoding from 'lib0/decoding';
+import * as encoding from 'lib0/encoding';
 import { Awareness } from 'y-protocols/awareness';
 import { WebsocketProvider as YWebsocketProvider } from 'y-websocket';
 
 import { requestDocSession } from './requests';
+import { MessageType, RoomMessage } from './utils';
 
 /**
  * An interface for a document provider.
@@ -111,6 +114,18 @@ export class WebSocketProvider implements IDocumentProvider {
 
     this._yWebsocketProvider.on('sync', this._onSync);
     this._yWebsocketProvider.on('connection-close', this._onConnectionClosed);
+
+    this._yWebsocketProvider.messageHandlers[MessageType.ROOM] = (
+      encoder,
+      decoder,
+      provider,
+      emitSynced,
+      messageType
+    ) => {
+      const msgType = decoding.readVarUint(decoder);
+      const data = decoding.readVarString(decoder);
+      this._handleRoomMessage(msgType, data);
+    };
   }
 
   private _onUserChanged(user: User.IManager): void {
@@ -138,6 +153,59 @@ export class WebSocketProvider implements IDocumentProvider {
     }
   };
 
+  private _handleRoomMessage(type: number, data: string): void {
+    switch (type) {
+      case RoomMessage.FILE_CHANGED:
+        this._handleFileChanged(data);
+        break;
+
+      case RoomMessage.DOC_OVERWRITTEN:
+      case RoomMessage.FILE_OVERWRITTEN:
+        if (this._dialog) {
+          this._dialog.close();
+          this._dialog = null;
+        }
+        break;
+    }
+  }
+
+  private _handleFileChanged(data: string): void {
+    this._dialog = new Dialog({
+      title: this._trans.__('File changed'),
+      body: this._trans.__('Do you want to overwrite the file or reload it?'),
+      buttons: [
+        Dialog.okButton({ label: 'Reload' }),
+        Dialog.warnButton({ label: 'Overwrite' })
+      ],
+      hasClose: false
+    });
+
+    this._dialog.launch().then(resp => {
+      if (resp.button.label === 'Reload') {
+        this._sendReloadMsg(data);
+      } else if (resp.button.label === 'Overwrite') {
+        this._sendOverwriteMsg(data);
+      }
+    });
+  }
+
+  private _sendReloadMsg(data: string): void {
+    const encoder = encoding.createEncoder();
+    encoding.writeVarUint(encoder, MessageType.ROOM);
+    encoding.writeVarUint(encoder, RoomMessage.RELOAD);
+    encoding.writeVarString(encoder, data);
+    this._yWebsocketProvider?.ws!.send(encoding.toUint8Array(encoder));
+  }
+
+  private _sendOverwriteMsg(data: string): void {
+    const encoder = encoding.createEncoder();
+    encoding.writeVarUint(encoder, MessageType.ROOM);
+    encoding.writeVarUint(encoder, RoomMessage.OVERWRITE);
+    encoding.writeVarString(encoder, data);
+    this._yWebsocketProvider?.ws!.send(encoding.toUint8Array(encoder));
+  }
+
+  private _dialog: Dialog<any> | null = null;
   private _awareness: Awareness;
   private _contentType: string;
   private _format: string;