refactor(realtime): connection flow to queue

Signed-off-by: BoHong Li <a60814billy@gmail.com>

Author: a60814billy

app.js

@@ -299,6 +299,9 @@ function handleTermSignals () {
       })
     }
   }, 100)
+  setTimeout(() => {
+    process.exit(1)
+  }, 5000)
 }
 process.on('SIGINT', handleTermSignals)
 process.on('SIGTERM', handleTermSignals)

lib/processQueue.js

@@ -7,24 +7,41 @@ const EventEmitter = require('events').EventEmitter
  */
 
 const QueueEvent = {
-  Tick: 'Tick'
+  Tick: 'Tick',
+  Push: 'Push',
+  Finish: 'Finish'
 }
 
 class ProcessQueue extends EventEmitter {
-  constructor (maximumLength, triggerTimeInterval = 10) {
+  constructor ({
+    maximumLength = 500,
+    triggerTimeInterval = 5000,
+    // execute on push
+    proactiveMode = true,
+    // execute next work on finish
+    continuousMode = true
+  }) {
     super()
     this.max = maximumLength
     this.triggerTime = triggerTimeInterval
     this.taskMap = new Map()
     this.queue = []
     this.lock = false
 
-    this.on(QueueEvent.Tick, () => {
-      if (this.lock) return
-      this.lock = true
-      setImmediate(() => {
-        this.process()
-      })
+    this.on(QueueEvent.Tick, this.onEventProcessFunc.bind(this))
+    if (proactiveMode) {
+      this.on(QueueEvent.Push, this.onEventProcessFunc.bind(this))
+    }
+    if (continuousMode) {
+      this.on(QueueEvent.Finish, this.onEventProcessFunc.bind(this))
+    }
+  }
+
+  onEventProcessFunc () {
+    if (this.lock) return
+    this.lock = true
+    setImmediate(() => {
+      this.process()
     })
   }
 
@@ -62,7 +79,7 @@ class ProcessQueue extends EventEmitter {
     this.taskMap.set(id, true)
     this.queue.push(task)
     this.start()
-    this.emit(QueueEvent.Tick)
+    this.emit(QueueEvent.Push)
     return true
   }
 
@@ -79,7 +96,7 @@ class ProcessQueue extends EventEmitter {
     const finishTask = () => {
       this.lock = false
       setImmediate(() => {
-        this.emit(QueueEvent.Tick)
+        this.emit(QueueEvent.Finish)
       })
     }
     task.processingFunc().then(finishTask).catch(finishTask)