Collaborative renaming & moving of files (#10470)

* Fix renaming jupyterlab/jupyterlab#10327

* implement PR suggestion and add mode documentation

* implement suggestion - use enum instead of integer variables

jupyterlab/handlers/yjs_echo_ws.py

@@ -8,11 +8,24 @@ import time
 
 from tornado.ioloop import IOLoop
 from tornado.websocket import WebSocketHandler
+from enum import IntEnum
 
-acquireLockMessageType = 127
-releaseLockMessageType = 126
-requestInitializedContentMessageType = 125
-putInitializedContentMessageType = 124
+## The y-protocol defines messages types that just need to be propagated to all other peers.
+## Here, we define some additional messageTypes that the server can interpret.
+## Messages that the server can't interpret should be broadcasted to all other clients.
+
+class ServerMessageType(IntEnum):
+    # The client is asking for a lock. Should return a lock-identifier if one is available.
+    ACQUIRE_LOCK = 127
+    # The client is asking to release a lock to make it available to other users again.
+    RELEASE_LOCK = 126
+    # The client is asking to retrieve the initial state of the Yjs document. Return an empty buffer when nothing is available.
+    REQUEST_INITIALIZED_CONTENT = 125
+    # The client retrieved an empty "initial content" and generated the initial state of the document after acquiring a lock. Store this.
+    PUT_INITIALIZED_CONTENT = 124
+    # The client moved the document to a different location. After receiving this message, we make the current document available under a different url.
+    # The other clients are automatically notified of this change because the path is shared through the Yjs document as well.
+    RENAME_SESSION = 123
 
 class YjsRoom:
     def __init__(self):
@@ -32,35 +45,44 @@ class YJSEchoWS(WebSocketHandler):
         if room is None:
             room = YjsRoom()
             cls.rooms[self.room_id] = room
-        room.clients[self.id] = ( IOLoop.current(), self.hook_send_message )
+        room.clients[self.id] = ( IOLoop.current(), self.hook_send_message, self )
         # Send SyncStep1 message (based on y-protocols)
         self.write_message(bytes([0, 0, 1, 0]), binary=True)
 
     def on_message(self, message):
         #print("[YJSEchoWS]: message, ", message)
         cls = self.__class__
-        room = cls.rooms.get(self.room_id)
-        if message[0] == acquireLockMessageType: # tries to acquire lock
+        room_id = self.room_id
+        room = cls.rooms.get(room_id)
+        if message[0] == ServerMessageType.ACQUIRE_LOCK:
             now = int(time.time())
             if room.lock is None or now - room.lock > 15: # no lock or timeout
                 room.lock = now
                 # print('Acquired new lock: ', room.lock)
                 # return acquired lock
-                self.write_message(bytes([acquireLockMessageType]) + room.lock.to_bytes(4, byteorder = 'little'), binary=True)
-        elif message[0] == releaseLockMessageType:
+                self.write_message(bytes([ServerMessageType.ACQUIRE_LOCK]) + room.lock.to_bytes(4, byteorder = 'little'), binary=True)
+        elif message[0] == ServerMessageType.RELEASE_LOCK:
             releasedLock = int.from_bytes(message[1:], byteorder = 'little')
             # print("trying release lock: ", releasedLock)
             if room.lock == releasedLock:
                 # print('released lock: ', room.lock)
                 room.lock = None
-        elif message[0] == requestInitializedContentMessageType:
+        elif message[0] == ServerMessageType.REQUEST_INITIALIZED_CONTENT:
             # print("client requested initial content")
-            self.write_message(bytes([requestInitializedContentMessageType]) + room.content, binary=True)
-        elif message[0] == putInitializedContentMessageType:
+            self.write_message(bytes([ServerMessageType.REQUEST_INITIALIZED_CONTENT]) + room.content, binary=True)
+        elif message[0] == ServerMessageType.PUT_INITIALIZED_CONTENT:
             # print("client put initialized content")
             room.content = message[1:]
+        elif message[0] == ServerMessageType.RENAME_SESSION:
+            # We move the room to a different entry and also change the room_id property of each connected client
+            new_room_id = message[1:].decode("utf-8")
+            for client_id, (loop, hook_send_message, client) in room.clients.items() :
+                client.room_id = new_room_id
+            cls.rooms.pop(room_id)
+            cls.rooms[new_room_id] = room
+            # print("renamed room to " + new_room_id + ". Old room name was " + room_id)
         elif room:
-            for client_id, (loop, hook_send_message) in room.clients.items() :
+            for client_id, (loop, hook_send_message, client) in room.clients.items() :
                 if self.id != client_id :
                     loop.add_callback(hook_send_message, message)
 

packages/docprovider/src/mock.ts

@@ -16,4 +16,7 @@ export class ProviderMock implements IDocumentProvider {
   destroy(): void {
     /* nop */
   }
+  setPath(path: string): void {
+    /* nop */
+  }
 }

packages/docprovider/src/tokens.ts

@@ -35,6 +35,11 @@ export interface IDocumentProvider {
    */
   releaseLock(lock: number): void;
 
+  /**
+   * This should be called by the docregistry when the file has been renamed to update the websocket connection url
+   */
+  setPath(newPath: string): void;
+
   /**
    * Destroy the provider.
    */
@@ -59,7 +64,8 @@ export namespace IDocumentProviderFactory {
     /**
      * The name (id) of the room
      */
-    guid: string;
+    path: string;
+    contentType: string;
 
     /**
      * The YNotebook.

packages/docprovider/src/yprovider.ts

@@ -7,7 +7,7 @@ import * as decoding from 'lib0/decoding';
 import * as encoding from 'lib0/encoding';
 import { WebsocketProvider } from 'y-websocket';
 import * as Y from 'yjs';
-import { IDocumentProviderFactory } from './tokens';
+import { IDocumentProvider, IDocumentProviderFactory } from './tokens';
 import { getAnonymousUserName, getRandomColor } from './awareness';
 import * as env from 'lib0/environment';
 
@@ -17,17 +17,30 @@ import * as env from 'lib0/environment';
  * The user can specify their own user-name and user-color by adding urlparameters:
  *   ?username=Alice&usercolor=007007
  * wher usercolor must be a six-digit hexadicimal encoded RGB value without the hash token.
+ *
+ * We specify custom messages that the server can interpret. For reference please look in yjs_ws_server.
+ *
  */
-export class WebSocketProviderWithLocks extends WebsocketProvider {
+export class WebSocketProviderWithLocks
+  extends WebsocketProvider
+  implements IDocumentProvider {
   /**
    * Construct a new WebSocketProviderWithLocks
    *
    * @param options The instantiation options for a WebSocketProviderWithLocks
    */
   constructor(options: WebSocketProviderWithLocks.IOptions) {
-    super(options.url, options.guid, options.ymodel.ydoc, {
-      awareness: options.ymodel.awareness
-    });
+    super(
+      options.url,
+      options.contentType + ':' + options.path,
+      options.ymodel.ydoc,
+      {
+        awareness: options.ymodel.awareness
+      }
+    );
+    this._path = options.path;
+    this._contentType = options.contentType;
+    this._serverUrl = options.url;
     const color = '#' + env.getParam('--usercolor', getRandomColor().slice(1));
     const name = env.getParam('--username', getAnonymousUserName());
     const awareness = options.ymodel.awareness;
@@ -83,6 +96,29 @@ export class WebSocketProviderWithLocks extends WebsocketProvider {
     this.on('status', this._onConnectionStatus);
   }
 
+  setPath(newPath: string): void {
+    if (newPath !== this._path) {
+      this._path = newPath;
+      // The next time the provider connects, we should connect through a different server url
+      this.bcChannel =
+        this._serverUrl + '/' + this._contentType + ':' + this._path;
+      this.url = this.bcChannel;
+      const encoder = encoding.createEncoder();
+      encoding.write(encoder, 123);
+      // writing a utf8 string to the encoder
+      const escapedPath = unescape(
+        encodeURIComponent(this._contentType + ':' + newPath)
+      );
+      for (let i = 0; i < escapedPath.length; i++) {
+        encoding.write(
+          encoder,
+          /** @type {number} */ escapedPath.codePointAt(i)!
+        );
+      }
+      this._sendMessage(encoding.toUint8Array(encoder));
+    }
+  }
+
   /**
    * Resolves to true if the initial content has been initialized on the server. false otherwise.
    */
@@ -198,6 +234,9 @@ export class WebSocketProviderWithLocks extends WebsocketProvider {
     }
   }
 
+  private _path: string;
+  private _contentType: string;
+  private _serverUrl: string;
   private _isInitialized: boolean;
   private _currentLockRequest: {
     promise: Promise<number>;

packages/docregistry/src/context.ts

@@ -71,10 +71,13 @@ export class Context<
     const ydoc = ymodel.ydoc;
     this._ydoc = ydoc;
     this._ycontext = ydoc.getMap('context');
-    const guid = this._factory.contentType + ':' + localPath;
     const docProviderFactory = options.docProviderFactory;
     this._provider = docProviderFactory
-      ? docProviderFactory({ guid, ymodel })
+      ? docProviderFactory({
+          path: this._path,
+          contentType: this._factory.contentType,
+          ymodel
+        })
       : new ProviderMock();
 
     this._readyPromise = manager.ready.then(() => {
@@ -100,6 +103,20 @@ export class Context<
     }));
     this.pathChanged.connect((sender, newPath) => {
       urlResolver.path = newPath;
+      if (this._ycontext.get('path') !== newPath) {
+        this._ycontext.set('path', newPath);
+      }
+    });
+    this._ycontext.set('path', this._path);
+    this._ycontext.observe(event => {
+      const pathChanged = event.changes.keys.get('path');
+      if (pathChanged) {
+        const newPath = this._ycontext.get('path')!;
+        this._provider.setPath(newPath);
+        if (newPath && newPath !== this.path) {
+          this.sessionContext.session?.setPath(newPath) as any;
+        }
+      }
     });
   }