Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: Implemented possibility to customize listener functions and import useful functions externally #191

Open
wants to merge 4 commits into
base: master
Choose a base branch
from
Open

feat: Implemented possibility to customize listener functions and import useful functions externally #191

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b455eb3
Implemented possibility to customize the message listener with `setMe…
carlosleandr Feb 25, 2025
1e0432e
Implemented possibility to customize the message sending function wit…
carlosleandr Feb 25, 2025
b776d4a
Implemented ability to customize the `updateHandler` function with `s…
carlosleandr Feb 26, 2025
b4a00b1
Update utils.cjs
carlosleandr Mar 10, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 55 additions & 6 deletions bin/utils.cjs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,14 +77,29 @@ const messageAwareness = 1
* @param {WSSharedDoc} doc
* @param {any} _tr
*/
const updateHandler = (update, _origin, doc, _tr) => {
exports.updateHandler = (update, _origin, doc, _tr) => {
const encoder = encoding.createEncoder()
encoding.writeVarUint(encoder, messageSync)
syncProtocol.writeUpdate(encoder, update)
const message = encoding.toUint8Array(encoder)
doc.conns.forEach((_, conn) => send(doc, conn, message))
}

/**
* @param {Uint8Array} update
* @param {any} _origin
* @param {WSSharedDoc} doc
* @param {any} _tr
*/
let updateHandler = exports.updateHandler;

/**
* @param {(update: Uint8Array, _origin: any, doc: WSSharedDoc, _tr: any) => Promise<void>} f
*/
exports.setUpdateHandler = (f) => {
updateHandler = f;
}

/**
* @type {(ydoc: Y.Doc) => Promise<void>}
*/
Expand Down Expand Up @@ -176,12 +191,12 @@ exports.getYDoc = getYDoc
/**
* @param {any} conn
* @param {WSSharedDoc} doc
* @param {Uint8Array} message
* @param {ArrayBuffer} message
*/
const messageListener = (conn, doc, message) => {
exports.messageListener = (conn, doc, message) => {
try {
const encoder = encoding.createEncoder()
const decoder = decoding.createDecoder(message)
const decoder = decoding.createDecoder(new Uint8Array(message))
const messageType = decoding.readVarUint(decoder)
switch (messageType) {
case messageSync:
Expand All @@ -207,6 +222,20 @@ const messageListener = (conn, doc, message) => {
}
}

/**
* @param {any} conn
* @param {WSSharedDoc} doc
* @param {ArrayBuffer} message
*/
let messageListener = exports.messageListener;

/**
* @param {(conn: any, doc: Y.Doc, message: ArrayBuffer) => void} f
*/
exports.setMessageListener = (f) => {
messageListener = f;
}

/**
* @param {WSSharedDoc} doc
* @param {any} conn
Expand All @@ -231,12 +260,18 @@ const closeConn = (doc, conn) => {
conn.close()
}

/**
* @param {WSSharedDoc} doc
* @param {any} conn
*/
exports.closeConn = closeConn;

/**
* @param {WSSharedDoc} doc
* @param {import('ws').WebSocket} conn
* @param {Uint8Array} m
*/
const send = (doc, conn, m) => {
exports.send = (doc, conn, m) => {
if (conn.readyState !== wsReadyStateConnecting && conn.readyState !== wsReadyStateOpen) {
closeConn(doc, conn)
}
Expand All @@ -247,6 +282,20 @@ const send = (doc, conn, m) => {
}
}

/**
* @param {WSSharedDoc} doc
* @param {import('ws').WebSocket} conn
* @param {Uint8Array} m
*/
let send = exports.send;

/**
* @param {(doc: WSSharedDoc, conn: WebSocket, m: Uint8Array) => void} f
*/
exports.setSend = (f) => {
send = f;
}

const pingTimeout = 30000

/**
Expand All @@ -260,7 +309,7 @@ exports.setupWSConnection = (conn, req, { docName = (req.url || '').slice(1).spl
const doc = getYDoc(docName, gc)
doc.conns.set(conn, new Set())
// listen and reply to events
conn.on('message', /** @param {ArrayBuffer} message */ message => messageListener(conn, doc, new Uint8Array(message)))
conn.on('message', /** @param {ArrayBuffer} message */ message => messageListener(conn, doc, message))

// Check if connection is still alive
let pongReceived = true
Expand Down