feat: add enableAwareness option to API and SocketIO server

Author: EastSun5566

src/api.js

@@ -84,10 +84,10 @@ const decodeRedisRoomStreamName = (rediskey, expectedPrefix) => {
 
 /**
  * @param {import('./storage.js').AbstractStorage} store
- * @param {{ redisPrefix?: string, redisUrl?: string }} opts
+ * @param {{ redisPrefix?: string, redisUrl?: string, enableAwareness?: boolean }} opts
  */
-export const createApiClient = async (store, { redisPrefix, redisUrl }) => {
-  const a = new Api(store, redisPrefix, redisUrl)
+export const createApiClient = async (store, { redisPrefix, redisUrl, enableAwareness = true }) => {
+  const a = new Api(store, redisPrefix, redisUrl, { enableAwareness })
   await a.redis.connect()
   try {
     await a.redis.xGroupCreate(a.redisWorkerStreamName, a.redisWorkerGroupName, '0', { MKSTREAM: true })
@@ -100,10 +100,13 @@ export class Api {
    * @param {import('./storage.js').AbstractStorage} store
    * @param {string=} prefix
    * @param {string=} url
+   * @param {Object} opts
+   * @param {boolean=} opts.enableAwareness
    */
-  constructor (store, prefix = 'y', url = env.ensureConf('ysr-redis')) {
+  constructor (store, prefix = 'y', url = env.ensureConf('ysr-redis'), { enableAwareness = true } = {}) {
     this.store = store
     this.prefix = prefix
+    this.enableAwareness = enableAwareness
     this.consumername = random.uuidv4()
     /**
      * After this timeout, a new worker will pick up the task
@@ -240,8 +243,11 @@ export class Api {
     if (docMessages?.messages) logApi(`processing messages of length: ${docMessages?.messages.length} in room: ${room}`)
     const docstate = await this.store.retrieveDoc(room, docid)
     const ydoc = new Y.Doc()
-    const awareness = new awarenessProtocol.Awareness(ydoc)
-    awareness.setLocalState(null) // we don't want to propagate awareness state
+    let awareness = null
+    if (this.enableAwareness) {
+      awareness = new awarenessProtocol.Awareness(ydoc)
+      awareness.setLocalState(null) // we don't want to propagate awareness state
+    }
     const now = performance.now()
     if (docstate) { Y.applyUpdateV2(ydoc, docstate.doc) }
     let changed = false
@@ -257,7 +263,9 @@ export class Api {
             break
           }
           case 1: { // awareness message
-            awarenessProtocol.applyAwarenessUpdate(awareness, decoding.readVarUint8Array(decoder), null)
+            if (this.enableAwareness && awareness) {
+              awarenessProtocol.applyAwarenessUpdate(awareness, decoding.readVarUint8Array(decoder), null)
+            }
             break
           }
         }
@@ -394,7 +402,7 @@ export class Api {
 
 /**
  * @param {import('./storage.js').AbstractStorage} store
- * @param {{ redisPrefix?: string, redisUrl?: string }} opts
+ * @param {{ redisPrefix?: string, redisUrl?: string, enableAwareness?: boolean }} opts
  */
 export const createWorker = async (store, opts) => {
   const a = await createApiClient(store, opts)

src/socketio.js

@@ -38,9 +38,16 @@ class YSocketIOServer {
  * @param {string} [conf.redisUrl]
  * @param {import('./y-socket-io/y-socket-io.js').YSocketIOConfiguration['authenticate']} conf.authenticate
  * @param {import('worker_threads').Worker=} [conf.persistWorker]
+ * @param {boolean} [conf.enableAwareness]
  */
-export const registerYSocketIOServer = async (io, store, { authenticate, redisUrl, redisPrefix, persistWorker }) => {
-  const app = new YSocketIO(io, { authenticate })
+export const registerYSocketIOServer = async (io, store, {
+  authenticate,
+  redisUrl,
+  redisPrefix,
+  persistWorker,
+  enableAwareness = true
+}) => {
+  const app = new YSocketIO(io, { authenticate, enableAwareness })
   const { client, subscriber } = await app.initialize(store, { redisUrl, redisPrefix, persistWorker })
   return new YSocketIOServer(app, client, subscriber)
 }

src/subscriber.js

@@ -32,7 +32,7 @@ const run = async subscriber => {
 
 /**
  * @param {import('./storage.js').AbstractStorage} store
- * @param {{ redisPrefix?: string, redisUrl?: string }} opts
+ * @param {{ redisPrefix?: string, redisUrl?: string, enableAwareness?: boolean }} opts
  */
 export const createSubscriber = async (store, opts) => {
   const client = await api.createApiClient(store, opts)

src/y-socket-io/y-socket-io.js

@@ -57,6 +57,8 @@ process.on('SIGINT', function () {
  *
  * @prop {(socket: Socket)=> Promise<UserLike | null> | UserLike | null} authenticate
  * Callback to authenticate the client connection.
+ * @prop {boolean=} enableAwareness
+ * Enable/disable awareness functionality, defaults to true
  */
 
 /**
@@ -159,7 +161,10 @@ export class YSocketIO {
    */
   constructor (io, configuration) {
     this.io = io
-    this.configuration = configuration
+    this.configuration = {
+      enableAwareness: true,
+      ...configuration
+    }
   }
 
   /**
@@ -174,9 +179,10 @@ export class YSocketIO {
    * @public
    */
   async initialize (store, { redisUrl, redisPrefix = 'y', persistWorker } = {}) {
+    const { enableAwareness } = this.configuration
     const [client, subscriber] = await promise.all([
-      api.createApiClient(store, { redisUrl, redisPrefix }),
-      createSubscriber(store, { redisUrl, redisPrefix })
+      api.createApiClient(store, { redisUrl, redisPrefix, enableAwareness }),
+      createSubscriber(store, { redisUrl, redisPrefix, enableAwareness })
     ])
     this.client = client
     this.subscriber = subscriber
@@ -243,7 +249,9 @@ export class YSocketIO {
       this.streamNamespaceMap.set(stream, namespace)
 
       this.initSyncListeners(socket)
-      this.initAwarenessListeners(socket)
+      if (this.configuration.enableAwareness) {
+        this.initAwarenessListeners(socket)
+      }
       this.initSocketListeners(socket)
       ;(async () => {
         assert(this.client)
@@ -404,7 +412,7 @@ export class YSocketIO {
           .catch(console.error)
       }
     )
-    if (doc.awareness.states.size > 0) {
+    if (this.configuration.enableAwareness && doc.awareness.states.size > 0) {
       socket.emit(
         'awareness-update',
         AwarenessProtocol.encodeAwarenessUpdate(
@@ -436,7 +444,7 @@ export class YSocketIO {
 
     for (const m of messages) {
       const decoded = this.fromRedis(m)
-      if (decoded.type === 'awareness-update') awareness.push(decoded.message)
+      if (decoded.type === 'awareness-update' && this.configuration.enableAwareness) awareness.push(decoded.message)
       else updates.push(decoded.message)
     }