Couch object provider performance improvement using SharedWorker (#3993)

* Use the window SharedWorker instead of the WorkerService
* Use relative asset path for Shared Workers
* Remove beforeunload listener on destroy

src/MCT.js

@@ -122,6 +122,7 @@ define([
             }
         };
 
+        this.destroy = this.destroy.bind(this);
         /**
          * Tracks current selection state of the application.
          * @private
@@ -435,6 +436,8 @@ define([
                     Browse(this);
                 }
 
+                window.addEventListener('beforeunload', this.destroy);
+
                 this.router.start();
                 this.emit('start');
             }.bind(this));
@@ -458,6 +461,7 @@ define([
     };
 
     MCT.prototype.destroy = function () {
+        window.removeEventListener('beforeunload', this.destroy);
         this.emit('destroy');
         this.router.destroy();
     };

src/plugins/persistence/couch/CouchChangesFeed.js

@@ -0,0 +1,106 @@
+(function () {
+    const connections = [];
+    let connected = false;
+    const controller = new AbortController();
+    const signal = controller.signal;
+
+    self.onconnect = function (e) {
+        let port = e.ports[0];
+        connections.push(port);
+
+        port.postMessage({
+            type: 'connection',
+            connectionId: connections.length
+        });
+
+        port.onmessage = async function (event) {
+            if (event.data.request === 'close') {
+                connections.splice(event.data.connectionId - 1, 1);
+                if (connections.length <= 0) {
+                    // abort any outstanding requests if there's nobody listening to it.
+                    controller.abort();
+                }
+
+                return;
+            }
+
+            if (event.data.request === 'changes') {
+                if (connected === true) {
+                    return;
+                }
+
+                connected = true;
+
+                let url = event.data.url;
+                let body = event.data.body;
+                let error = false;
+                // feed=continuous maintains an indefinitely open connection with a keep-alive of HEARTBEAT milliseconds until this client closes the connection
+                // style=main_only returns only the current winning revision of the document
+
+                const response = await fetch(url, {
+                    method: 'POST',
+                    headers: {
+                        "Content-Type": 'application/json'
+                    },
+                    signal,
+                    body
+                });
+
+                let reader;
+
+                if (response.body === undefined) {
+                    error = true;
+                } else {
+                    reader = response.body.getReader();
+                }
+
+                while (!error) {
+                    const {done, value} = await reader.read();
+                    //done is true when we lose connection with the provider
+                    if (done) {
+                        error = true;
+                    }
+
+                    if (value) {
+                        let chunk = new Uint8Array(value.length);
+                        chunk.set(value, 0);
+                        const decodedChunk = new TextDecoder("utf-8").decode(chunk).split('\n');
+                        if (decodedChunk.length && decodedChunk[decodedChunk.length - 1] === '') {
+                            decodedChunk.forEach((doc, index) => {
+                                try {
+                                    if (doc) {
+                                        const objectChanges = JSON.parse(doc);
+                                        connections.forEach(function (connection) {
+                                            connection.postMessage({
+                                                objectChanges
+                                            });
+                                        });
+                                    }
+                                } catch (decodeError) {
+                                    //do nothing;
+                                    console.log(decodeError);
+                                }
+                            });
+                        }
+                    }
+
+                }
+
+                if (error) {
+                    port.postMessage({
+                        error
+                    });
+                }
+            }
+        };
+
+        port.start();
+
+    };
+
+    self.onerror = function () {
+        //do nothing
+        console.log('Error on feed');
+    };
+
+}());

src/plugins/persistence/couch/CouchObjectProvider.js

@@ -40,6 +40,64 @@ export default class CouchObjectProvider {
         this.batchIds = [];
     }
 
+    /**
+     * @private
+     */
+    startSharedWorker() {
+        let provider = this;
+        let sharedWorker;
+
+        const sharedWorkerURL = `${this.openmct.getAssetPath()}${__OPENMCT_ROOT_RELATIVE__}couchDBChangesFeed.js`;
+
+        sharedWorker = new SharedWorker(sharedWorkerURL);
+        sharedWorker.port.onmessage = provider.onSharedWorkerMessage.bind(this);
+        sharedWorker.port.onmessageerror = provider.onSharedWorkerMessageError.bind(this);
+        sharedWorker.port.start();
+
+        this.openmct.on('destroy', () => {
+            this.changesFeedSharedWorker.port.postMessage({
+                request: 'close',
+                connectionId: this.changesFeedSharedWorkerConnectionId
+            });
+            this.changesFeedSharedWorker.port.close();
+        });
+
+        return sharedWorker;
+    }
+
+    onSharedWorkerMessageError(event) {
+        console.log('Error', event);
+    }
+
+    onSharedWorkerMessage(event) {
+        if (event.data.type === 'connection') {
+            this.changesFeedSharedWorkerConnectionId = event.data.connectionId;
+        } else {
+            const error = event.data.error;
+            if (error && Object.keys(this.observers).length > 0) {
+                this.observeObjectChanges();
+
+                return;
+            }
+
+            let objectChanges = event.data.objectChanges;
+            objectChanges.identifier = {
+                namespace: this.namespace,
+                key: objectChanges.id
+            };
+            let keyString = this.openmct.objects.makeKeyString(objectChanges.identifier);
+            //TODO: Optimize this so that we don't 'get' the object if it's current revision (from this.objectQueue) is the same as the one we already have.
+            let observersForObject = this.observers[keyString];
+
+            if (observersForObject) {
+                observersForObject.forEach(async (observer) => {
+                    const updatedObject = await this.get(objectChanges.identifier);
+                    observer(updatedObject);
+                });
+            }
+        }
+    }
+
     //backwards compatibility, options used to be a url. Now it's an object
     _normalize(options) {
         if (typeof options === 'string') {
@@ -334,9 +392,8 @@ export default class CouchObjectProvider {
     /**
      * @private
      */
-    async observeObjectChanges() {
-        const controller = new AbortController();
-        const signal = controller.signal;
+    observeObjectChanges() {
+
         let filter = {selector: {}};
 
         if (this.openmct.objects.SYNCHRONIZED_OBJECT_TYPES.length > 1) {
@@ -354,6 +411,51 @@ export default class CouchObjectProvider {
             };
         }
 
+        // feed=continuous maintains an indefinitely open connection with a keep-alive of HEARTBEAT milliseconds until this client closes the connection
+        // style=main_only returns only the current winning revision of the document
+        let url = `${this.url}/_changes?feed=continuous&style=main_only&heartbeat=${HEARTBEAT}`;
+
+        let body = {};
+        if (filter) {
+            url = `${url}&filter=_selector`;
+            body = JSON.stringify(filter);
+        }
+
+        if (typeof SharedWorker === 'undefined') {
+            this.fetchChanges(url, body);
+        } else {
+            this.initiateSharedWorkerFetchChanges(url, body);
+        }
+
+    }
+
+    /**
+     * @private
+     */
+    initiateSharedWorkerFetchChanges(url, body) {
+        if (!this.changesFeedSharedWorker) {
+            this.changesFeedSharedWorker = this.startSharedWorker();
+
+            if (typeof this.stopObservingObjectChanges === 'function') {
+                this.stopObservingObjectChanges();
+            }
+
+            this.stopObservingObjectChanges = () => {
+                delete this.stopObservingObjectChanges;
+            };
+
+            this.changesFeedSharedWorker.port.postMessage({
+                request: 'changes',
+                body,
+                url
+            });
+        }
+    }
+
+    async fetchChanges(url, body) {
+        const controller = new AbortController();
+        const signal = controller.signal;
+
         let error = false;
 
         if (typeof this.stopObservingObjectChanges === 'function') {
@@ -365,16 +467,6 @@ export default class CouchObjectProvider {
             delete this.stopObservingObjectChanges;
         };
 
-        // feed=continuous maintains an indefinitely open connection with a keep-alive of HEARTBEAT milliseconds until this client closes the connection
-        // style=main_only returns only the current winning revision of the document
-        let url = `${this.url}/_changes?feed=continuous&style=main_only&heartbeat=${HEARTBEAT}`;
-
-        let body = {};
-        if (filter) {
-            url = `${url}&filter=_selector`;
-            body = JSON.stringify(filter);
-        }
-
         const response = await fetch(url, {
             method: 'POST',
             signal,

webpack.config.js

@@ -14,19 +14,21 @@ const gitRevision = require('child_process')
 const gitBranch = require('child_process')
     .execSync('git rev-parse --abbrev-ref HEAD')
     .toString().trim();
-const vueFile = devMode ?
-    path.join(__dirname, "node_modules/vue/dist/vue.js") :
-    path.join(__dirname, "node_modules/vue/dist/vue.min.js");
+const vueFile = devMode
+    ? path.join(__dirname, "node_modules/vue/dist/vue.js")
+    : path.join(__dirname, "node_modules/vue/dist/vue.min.js");
 
 const webpackConfig = {
     mode: devMode ? 'development' : 'production',
     entry: {
         openmct: './openmct.js',
+        couchDBChangesFeed: './src/plugins/persistence/couch/CouchChangesFeed.js',
         espressoTheme: './src/plugins/themes/espresso-theme.scss',
         snowTheme: './src/plugins/themes/snow-theme.scss',
         maelstromTheme: './src/plugins/themes/maelstrom-theme.scss'
     },
     output: {
+        globalObject: "this",
         filename: '[name].js',
         library: '[name]',
         libraryTarget: 'umd',
@@ -105,13 +107,15 @@ const webpackConfig = {
                     name: '[name].[ext]',
                     outputPath(url, resourcePath, context) {
                         if (/\.(jpg|jpeg|png|svg)$/.test(url)) {
-                            return `images/${url}`
+                            return `images/${url}`;
                         }
+
                         if (/\.ico$/.test(url)) {
-                            return `icons/${url}`
+                            return `icons/${url}`;
                         }
+
                         if (/\.(woff2?|eot|ttf)$/.test(url)) {
-                            return `fonts/${url}`
+                            return `fonts/${url}`;
                         } else {
                             return `${url}`;
                         }