feat: move docs map to docroom #83
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -24,9 +24,6 @@ const wsReadyStateOpen = 1; | |
| // disable gc when using snapshots! | ||
| const gcEnabled = false; | ||
|
|
||
| const messageSync = 0; | ||
| const messageAwareness = 1; | ||
| const MAX_STORAGE_KEYS = 128; | ||
|
|
@@ -39,14 +36,12 @@ const MAX_STORAGE_VALUE_SIZE = 131072; | |
| * @param {WebSocket} conn - the websocket connection to close. | ||
| */ | ||
| export const closeConn = (doc, conn) => { | ||
| // eslint-disable-next-line no-console | ||
| console.log('Closing connection for - removing awareness states', doc.name); | ||
| if (doc.conns.has(conn)) { | ||
| const controlledIds = doc.conns.get(conn); | ||
| doc.conns.delete(conn); | ||
| awarenessProtocol.removeAwarenessStates(doc.awareness, Array.from(controlledIds), null); | ||
| } | ||
| conn.close(); | ||
| }; | ||
|
|
@@ -168,7 +163,7 @@ export const showError = (ydoc, err) => { | |
| }; | ||
|
|
||
| export const persistence = { | ||
| closeConn, | ||
|
|
||
| /** | ||
| * Get the document from da-admin. If da-admin doesn't have the doc, a new empty doc is | ||
|
|
@@ -278,7 +273,7 @@ export const persistence = { | |
| * @param {WebSocket} conn - the websocket connection | ||
| * @param {TransactionalStorage} storage - the worker transactional storage object | ||
| */ | ||
| bindState: async (docName, ydoc, conn, storage, docsCache) => { | ||
| let timingReadStateDuration; | ||
| let timingDaAdminGetDuration; | ||
|
|
||
|
|
@@ -337,7 +332,7 @@ export const persistence = { | |
| // eslint-disable-next-line no-console | ||
| console.log('Could not be restored, trying to restore from da-admin', docName); | ||
| setTimeout(() => { | ||
| if (ydoc === docsCache.get(docName)) { | ||
| const rootType = ydoc.getXmlFragment('prosemirror'); | ||
| ydoc.transact(() => { | ||
| try { | ||
|
|
@@ -360,15 +355,15 @@ export const persistence = { | |
|
|
||
| ydoc.on('update', async () => { | ||
| // Whenever we receive an update on the document store it in the local storage | ||
| if (ydoc === docsCache.get(docName)) { // make sure this ydoc is still active | ||
| storeState(docName, Y.encodeStateAsUpdate(ydoc), storage); | ||
| } | ||
| }); | ||
|
|
||
| ydoc.on('update', debounce(async () => { | ||
| // If we receive an update on the document, store it in da-admin, but debounce it | ||
| // to avoid excessive da-admin calls. | ||
| if (ydoc === docsCache.get(docName)) { | ||
| current = await persistence.update(ydoc, current); | ||
| } | ||
| }, 2000, { maxWait: 10000 })); | ||
|
|
@@ -436,25 +431,20 @@ export class WSSharedDoc extends Y.Doc { | |
| * @param {boolean} gc - whether garbage collection is enabled | ||
| * @returns The Yjs document object, which may be shared across multiple sockets. | ||
| */ | ||
| export const createYDoc = (docname, conn, env, storage, docsCache, gc = true) => { | ||
| const doc = new WSSharedDoc(docname); | ||
| doc.gc = gc; | ||
|
|
||
| // Store the service binding to da-admin which we receive through the environment in the doc | ||
| doc.daadmin = env.daadmin; | ||
| doc.promise = persistence.bindState(docname, doc, conn, storage, docsCache); | ||
|
Collaborator
There was a problem hiding this comment. I think this is not doing what happened before. IIRC, the idea was to use the promise to make sure the doc doesn't get bound to persistence called concurrently. @bosschaert, are you sure this is still correct (I would have expected to have this move up to the caller like it used to for the await)...
Contributor
There was a problem hiding this comment. Ah yes, the
Contributor
Author
There was a problem hiding this comment. ok, I restored the logic but split the method in 2: |
||
|
|
||
| return doc; | ||
| }; | ||
|
|
||
| export const setupYDoc = async (doc, conn, timingData) => { | ||
| if (!doc.conns.get(conn)) { | ||
| doc.conns.set(conn, new Set()); | ||
| } | ||
|
|
||
| // We wait for the promise, for second and subsequent connections to the same doc, this will | ||
|
|
@@ -463,12 +453,8 @@ export const getYDoc = async (docname, conn, env, storage, timingData, gc = true | |
| if (timingData) { | ||
| timings.forEach((v, k) => timingData.set(k, v)); | ||
| } | ||
| }; | ||
|
|
||
| // This read sync message handles readonly connections | ||
| const readSyncMessage = (decoder, encoder, doc, readOnly, transactionOrigin) => { | ||
| const messageType = decoding.readVarUint(decoder); | ||
|
|
@@ -489,10 +475,11 @@ const readSyncMessage = (decoder, encoder, doc, readOnly, transactionOrigin) => | |
| }; | ||
|
|
||
| export const messageListener = (conn, doc, message) => { | ||
| let messageType; | ||
| try { | ||
| const encoder = encoding.createEncoder(); | ||
| const decoder = decoding.createDecoder(message); | ||
| messageType = decoding.readVarUint(decoder); | ||
| switch (messageType) { | ||
| case messageSync: | ||
| encoding.writeVarUint(encoder, messageSync); | ||
|
|
@@ -515,7 +502,7 @@ export const messageListener = (conn, doc, message) => { | |
| } | ||
| } catch (err) { | ||
| // eslint-disable-next-line no-console | ||
| console.error(`Error in messageListener ${doc?.name} - messageType: ${messageType}`, err); | ||
| showError(doc, err); | ||
| } | ||
| }; | ||
|
|
@@ -528,18 +515,18 @@ export const messageListener = (conn, doc, message) => { | |
| * @param {string} docName - The name of the document | ||
| * @returns true if the document was found and invalidated, false otherwise. | ||
| */ | ||
| export const invalidateFromAdmin = async (ydoc) => { | ||
| if (ydoc) { | ||
| // eslint-disable-next-line no-console | ||
| console.log('Invalidating document', ydoc.name); | ||
|
|
||
| // As we are closing all connections, the ydoc will be removed from the docs map | ||
| ydoc.conns.forEach((_, c) => closeConn(ydoc, c)); | ||
|
|
||
| return true; | ||
| } else { | ||
| // eslint-disable-next-line no-console | ||
| console.log('No document to invalidate'); | ||
| } | ||
| return false; | ||
| }; | ||
|
|
@@ -552,38 +539,32 @@ export const invalidateFromAdmin = async (docName) => { | |
| * @param {TransactionalStorage} storage - The worker transactional storage object | ||
| * @returns {Promise<void>} - The return value of this | ||
| */ | ||
| export const setupWSConnection = async (conn, ydoc) => { | ||
| // eslint-disable-next-line no-param-reassign | ||
| conn.binaryType = 'arraybuffer'; | ||
|
|
||
| // listen and reply to events | ||
| conn.addEventListener('message', (message) => messageListener(conn, ydoc, new Uint8Array(message.data))); | ||
|
|
||
| // Check if connection is still alive | ||
| conn.addEventListener('close', () => { | ||
| closeConn(ydoc, conn); | ||
| }); | ||
| // put the following in a variables in a block so the interval handlers don't keep in in | ||
| // scope | ||
| { | ||
| // send sync step 1 | ||
| let encoder = encoding.createEncoder(); | ||
| encoding.writeVarUint(encoder, messageSync); | ||
| syncProtocol.writeSyncStep1(encoder, ydoc); | ||
| send(ydoc, conn, encoding.toUint8Array(encoder)); | ||
| const awarenessStates = ydoc.awareness.getStates(); | ||
| if (awarenessStates.size > 0) { | ||
| encoder = encoding.createEncoder(); | ||
| encoding.writeVarUint(encoder, messageAwareness); | ||
| encoding.writeVarUint8Array(encoder, awarenessProtocol | ||
| .encodeAwarenessUpdate(ydoc.awareness, Array.from(awarenessStates.keys()))); | ||
| send(ydoc, conn, encoding.toUint8Array(encoder)); | ||
| } | ||
| } | ||
| }; | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'm not sure this is correct. Wouldn't this basically remove the doc if a connection goes away?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
On connection close, the doc is removed from the map ("local cache"). This is the same logic as before, just moved the code here to have access to the cache.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ok, what was missing in the new version is the check if there is no more connections. Added.