Implement http based callback

bin/callback.js

@@ -0,0 +1,84 @@
+const http = require('http')
+
+const CALLBACK_URL = process.env.CALLBACK_URL ? new URL(process.env.CALLBACK_URL) : null
+const CALLBACK_TIMEOUT = process.env.CALLBACK_TIMEOUT || 5000
+const CALLBACK_OBJECTS = process.env.CALLBACK_OBJECTS ? JSON.parse(process.env.CALLBACK_OBJECTS) : {}
+
+/**
+ * @return {Boolean}
+ */
+exports.isCallbackSet = () => {
+  if (CALLBACK_URL) {
+    return true
+  }
+  return false
+}
+
+/**
+ * @param {Uint8Array} update
+ * @param {any} origin
+ * @param {WSSharedDoc} doc
+ */
+exports.callbackHandler = (update, origin, doc) => {
+  const room = doc.name
+  const dataToSend = {
+    room: room,
+    data: {}
+  }
+  const sharedObjectList = Object.keys(CALLBACK_OBJECTS)
+  sharedObjectList.forEach(sharedObjectName => {
+    const sharedObjectType = CALLBACK_OBJECTS[sharedObjectName]
+    dataToSend.data[sharedObjectName] = {
+      type: sharedObjectType,
+      content: getContent(sharedObjectName, sharedObjectType, doc)
+    }
+  })
+  callbackRequest(CALLBACK_URL, CALLBACK_TIMEOUT, dataToSend)
+}
+
+/**
+ * @param {URL} url
+ * @param {number} timeout
+ * @param {Object} data
+ */
+const callbackRequest = (url, timeout, data) => {
+  data = JSON.stringify(data)
+  const options = {
+    hostname: url.hostname,
+    port: url.port,
+    path: url.pathname,
+    timeout: timeout,
+    method: 'POST',
+    headers: {
+      'Content-Type': 'application/json',
+      'Content-Length': data.length
+    }
+  }
+  var req = http.request(options)
+  req.on('timeout', () => {
+    console.warn('Callback request timed out.')
+    req.abort()
+  })
+  req.on('error', (e) => {
+    console.err('Callback request error.', e)
+    req.abort()
+  })
+  req.write(data)
+  req.end()
+}
+
+/**
+ * @param {string} objName
+ * @param {string} objType
+ * @param {WSSharedDoc} doc
+ */
+const getContent = (objName, objType, doc) => {
+  switch (objType) {
+    case 'Array': return doc.getArray(objName)
+    case 'Map': return doc.getMap(objName)
+    case 'Text': return doc.getText(objName)
+    case 'XmlFragment': return doc.getXmlFragment(objName)
+    case 'XmlElement': return doc.getXmlElement(objName)
+    default : return {}
+  }
+}

bin/utils.js

@@ -7,6 +7,14 @@ const decoding = require('lib0/dist/decoding.cjs')
 const mutex = require('lib0/dist/mutex.cjs')
 const map = require('lib0/dist/map.cjs')
 
+const debounce = require('lodash.debounce')
+
+const callbackHandler = require('./callback.js').callbackHandler
+const isCallbackSet = require('./callback.js').isCallbackSet
+
+const CALLBACK_DEBOUNCE_WAIT = process.env.CALLBACK_DEBOUNCE_WAIT || 2000
+const CALLBACK_DEBOUNCE_MAXWAIT = process.env.CALLBACK_DEBOUNCE_MAXWAIT || 10000
+
 const wsReadyStateConnecting = 0
 const wsReadyStateOpen = 1
 const wsReadyStateClosing = 2 // eslint-disable-line
@@ -110,6 +118,13 @@ class WSSharedDoc extends Y.Doc {
     }
     this.awareness.on('update', awarenessChangeHandler)
     this.on('update', updateHandler)
+    if (isCallbackSet()) {
+      this.on('update', debounce(
+        callbackHandler,
+        CALLBACK_DEBOUNCE_WAIT,
+        { maxWait: CALLBACK_DEBOUNCE_MAXWAIT }
+      ))
+    }
   }
 }
 

package-lock.json

@@ -1192,6 +1192,11 @@
       "integrity": "sha512-8xOcRHvCjnocdS5cpwXQXVzmmh5e5+saE2QGoeQmbKmRS6J3VQppPOIt0MnmE+4xlZoumy0GPG0D0MVIQbNA1A==",
       "dev": true
     },
+    "lodash.debounce": {
+      "version": "4.0.8",
+      "resolved": "https://registry.npmjs.org/lodash.debounce/-/lodash.debounce-4.0.8.tgz",
+      "integrity": "sha1-gteb/zCmfEAF/9XiUVMArZyk168="
+    },
     "loose-envify": {
       "version": "1.4.0",
       "resolved": "https://registry.npmjs.org/loose-envify/-/loose-envify-1.4.0.tgz",

package.json

@@ -47,6 +47,7 @@
   "dependencies": {
     "lib0": "^0.2.31",
     "y-leveldb": "^0.1.0",
+    "lodash.debounce": "^4.0.8",
     "y-protocols": "^1.0.0"
   },
   "devDependencies": {