refactor(realtime): disconnect flow

1. use queue for queueing disconnect request

Signed-off-by: BoHong Li <[email protected]>

Author: a60814billy

app.js

@@ -278,6 +278,7 @@ process.on('uncaughtException', function (err) {
 function handleTermSignals () {
   logger.info('CodiMD has been killed by signal, try to exit gracefully...')
   realtime.maintenance = true
+  realtime.terminate()
   // disconnect all socket.io clients
   Object.keys(io.sockets.sockets).forEach(function (key) {
     var socket = io.sockets.sockets[key]

lib/connectionQueue.js renamed to lib/processQueue.js

@@ -6,20 +6,22 @@ const EventEmitter = require('events').EventEmitter
  * Queuing Class for connection queuing
  */
 
-const ConnectionQueueEvent = {
+const QueueEvent = {
   Tick: 'Tick'
 }
 
-class ConnectionQueue extends EventEmitter {
+class ProcessQueue extends EventEmitter {
   constructor (maximumLength, triggerTimeInterval = 10) {
     super()
     this.max = maximumLength
     this.triggerTime = triggerTimeInterval
+    this.taskMap = new Map()
     this.queue = []
     this.lock = false
 
-    this.on(ConnectionQueueEvent.Tick, () => {
+    this.on(QueueEvent.Tick, () => {
       if (this.lock) return
+      this.lock = true
       setImmediate(() => {
         this.process()
       })
@@ -29,7 +31,7 @@ class ConnectionQueue extends EventEmitter {
   start () {
     if (this.eventTrigger) return
     this.eventTrigger = setInterval(() => {
-      this.emit(ConnectionQueueEvent.Tick)
+      this.emit(QueueEvent.Tick)
     }, this.triggerTime)
   }
 
@@ -40,36 +42,48 @@ class ConnectionQueue extends EventEmitter {
     }
   }
 
+  checkTaskIsInQueue (id) {
+    return this.taskMap.has(id)
+  }
+
   /**
-   * push a promisify-task to queue
-   * @param task {Promise}
-   * @returns {boolean} if success return true, otherwise flase
+   * pushWithKey a promisify-task to queue
+   * @param id {string}
+   * @param processingFunc {Function<Promise>}
+   * @returns {boolean} if success return true, otherwise false
    */
-  push (task) {
+  push (id, processingFunc) {
     if (this.queue.length >= this.max) return false
+    if (this.checkTaskIsInQueue(id)) return false
+    const task = {
+      id: id,
+      processingFunc: processingFunc
+    }
+    this.taskMap.set(id, true)
     this.queue.push(task)
     this.start()
+    this.emit(QueueEvent.Tick)
     return true
   }
 
   process () {
-    if (this.lock) return
-    this.lock = true
     if (this.queue.length <= 0) {
       this.stop()
       this.lock = false
       return
     }
+
     const task = this.queue.shift()
+    this.taskMap.delete(task.id)
 
     const finishTask = () => {
       this.lock = false
       setImmediate(() => {
-        this.process()
+        this.emit(QueueEvent.Tick)
       })
     }
-    task().then(finishTask).catch(finishTask)
+    task.processingFunc().then(finishTask).catch(finishTask)
   }
 }
 
-exports.ConnectionQueue = ConnectionQueue
+exports.ProcessQueue = ProcessQueue

lib/realtime.js

@@ -21,6 +21,7 @@ const models = require('./models')
 // ot
 const ot = require('./ot')
 
+const { ProcessQueue } = require('./processQueue')
 const { RealtimeClientConnection } = require('./realtimeClientConnection')
 const { UpdateDirtyNoteJob } = require('./realtimeUpdateDirtyNoteJob')
 
@@ -69,6 +70,7 @@ function secure (socket, next) {
   }
 }
 
+// TODO: only use in `updateDirtyNote`
 function emitCheck (note) {
   var out = {
     title: note.title,
@@ -85,6 +87,9 @@ function emitCheck (note) {
 var users = {}
 var notes = {}
 
+const disconnectProcessQueue = new ProcessQueue(2000, 500)
+disconnectProcessQueue.start()
+
 const updateDirtyNoteJob = new UpdateDirtyNoteJob(realtime)
 updateDirtyNoteJob.start(realtime)
 
@@ -173,8 +178,9 @@ setInterval(function () {
           id: key
         }
       }
-      disconnectSocketQueue.push(socket)
-      disconnect(socket)
+      if (!disconnectProcessQueue.checkTaskIsInQueue(socket.id)) {
+        exports.queueForDisconnect(socket)
+      }
     }
     return callback(null, null)
   }, function (err) {
@@ -238,8 +244,8 @@ function getStatus (callback) {
         distinctOnlineRegisteredUsers: distinctregaddresses.length,
         isConnectionBusy: isConnectionBusy,
         connectionSocketQueueLength: connectionSocketQueue.length,
-        isDisconnectBusy: isDisconnectBusy,
-        disconnectSocketQueueLength: disconnectSocketQueue.length
+        isDisconnectBusy: disconnectProcessQueue.lock,
+        disconnectSocketQueueLength: disconnectProcessQueue.queue.length
       }) : null
     }).catch(function (err) {
       return logger.error('count user failed: ' + err)
@@ -253,7 +259,7 @@ function isReady () {
   return realtime.io &&
     Object.keys(notes).length === 0 && Object.keys(users).length === 0 &&
     connectionSocketQueue.length === 0 && !isConnectionBusy &&
-    disconnectSocketQueue.length === 0 && !isDisconnectBusy
+    disconnectProcessQueue.queue.length === 0 && !disconnectProcessQueue.lock
 }
 
 function parseUrl (data) {
@@ -416,8 +422,6 @@ function checkViewPermission (req, note) {
 
 var isConnectionBusy = false
 var connectionSocketQueue = []
-var isDisconnectBusy = false
-var disconnectSocketQueue = []
 
 function finishConnection (socket, noteId, socketId) {
   // if no valid info provided will drop the client
@@ -562,71 +566,45 @@ function failConnection (code, err, socket) {
   return socket.disconnect(true)
 }
 
-function disconnect (socket) {
-  if (isDisconnectBusy) return
-  isDisconnectBusy = true
-
-  if (config.debug) {
-    logger.info('SERVER disconnected a client')
-    logger.info(JSON.stringify(users[socket.id]))
-  }
-
-  if (users[socket.id]) {
-    delete users[socket.id]
-  }
-  var noteId = socket.noteId
-  var note = notes[noteId]
-  if (note) {
-    // delete user in users
-    if (note.users[socket.id]) {
-      delete note.users[socket.id]
+function queueForDisconnect (socket) {
+  disconnectProcessQueue.push(socket.id, async function () {
+    if (users[socket.id]) {
+      delete users[socket.id]
     }
-    // remove sockets in the note socks
-    do {
-      var index = note.socks.indexOf(socket)
-      if (index !== -1) {
-        note.socks.splice(index, 1)
+    const noteId = socket.noteId
+    const note = notes[noteId]
+    if (note) {
+      // delete user in users
+      if (note.users[socket.id]) {
+        delete note.users[socket.id]
       }
-    } while (index !== -1)
-    // remove note in notes if no user inside
-    if (Object.keys(note.users).length <= 0) {
-      if (note.server.isDirty) {
-        updateNote(note, function (err, _note) {
-          if (err) return logger.error('disconnect note failed: ' + err)
-          // clear server before delete to avoid memory leaks
-          note.server.document = ''
-          note.server.operations = []
+      // remove sockets in the note socks
+      let index
+      do {
+        index = note.socks.indexOf(socket)
+        if (index !== -1) {
+          note.socks.splice(index, 1)
+        }
+      } while (index !== -1)
+      // remove note in notes if no user inside
+      if (Object.keys(note.users).length === 0) {
+        if (note.server.isDirty) {
+          exports.updateNote(note, function (err, _note) {
+            if (err) return logger.error('disconnect note failed: ' + err)
+            // clear server before delete to avoid memory leaks
+            note.server.document = ''
+            note.server.operations = []
+            delete note.server
+            delete notes[noteId]
+          })
+        } else {
           delete note.server
           delete notes[noteId]
-          if (config.debug) {
-            // logger.info(notes);
-            getStatus(function (data) {
-              logger.info(JSON.stringify(data))
-            })
-          }
-        })
-      } else {
-        delete note.server
-        delete notes[noteId]
+        }
       }
     }
-  }
-  emitOnlineUsers(socket)
-
-  // clear finished socket in queue
-  clearSocketQueue(disconnectSocketQueue, socket)
-  // seek for next socket
-  isDisconnectBusy = false
-  if (disconnectSocketQueue.length > 0) {
-    disconnect(disconnectSocketQueue[0])
-  }
-
-  if (config.debug) {
-    // logger.info(notes);
-    getStatus(function (data) {
-      logger.info(JSON.stringify(data))
-    })
-  }
+    exports.emitOnlineUsers(socket)
+  })
 }
 
 function buildUserOutData (user) {
@@ -818,6 +796,11 @@ function connection (socket) {
   socketClient.registerEventHandler()
 }
 
+function terminate () {
+  disconnectProcessQueue.stop()
+  updateDirtyNoteJob.stop()
+}
+
 exports = module.exports = realtime
 exports.extractNoteIdFromSocket = extractNoteIdFromSocket
 exports.parseNoteIdFromSocket = parseNoteIdFromSocket
@@ -829,7 +812,6 @@ exports.updateUserData = updateUserData
 exports.startConnection = startConnection
 exports.emitRefresh = emitRefresh
 exports.emitUserStatus = emitUserStatus
-exports.disconnect = disconnect
 exports.emitOnlineUsers = emitOnlineUsers
 exports.checkViewPermission = checkViewPermission
 exports.getNoteFromNotePool = getNoteFromNotePool
@@ -838,6 +820,9 @@ exports.buildUserOutData = buildUserOutData
 exports.getNotePool = getNotePool
 exports.emitCheck = emitCheck
 exports.disconnectSocketOnNote = disconnectSocketOnNote
+exports.queueForDisconnect = queueForDisconnect
+exports.terminate = terminate
+exports.getUserPool = getUserPool
+exports.disconnectProcessQueue = disconnectProcessQueue
 exports.notes = notes
 exports.users = users
-exports.disconnectSocketQueue = disconnectSocketQueue

lib/realtimeClientConnection.js

@@ -222,9 +222,10 @@ class RealtimeClientConnection {
   }
 
   disconnectEventHandler () {
-    if (this.realtime.isDuplicatedInSocketQueue(this.realtime.disconnectSocketQueue, this.socket)) return
-    this.realtime.disconnectSocketQueue.push(this.socket)
-    this.realtime.disconnect(this.socket)
+    if (this.realtime.disconnectProcessQueue.checkTaskIsInQueue(this.socket.id)) {
+      return
+    }
+    this.realtime.queueForDisconnect(this.socket)
   }
 }