Use Server-Sent Events in the Couch DB adapter (#4427)

* Use SSE instead of chunked HTTP response.

Co-authored-by: Joshi <simplyrender@gmail.com>

karma.conf.js

@@ -52,7 +52,11 @@ module.exports = (config) => {
         basePath: '',
         frameworks: ['jasmine'],
         files: [
-            'indexTest.js'
+            'indexTest.js',
+            {
+                pattern: 'dist/couchDBChangesFeed.js',
+                included: false
+            }
         ],
         port: 9876,
         reporters: reporters,

src/plugins/persistence/couch/CouchChangesFeed.js

@@ -1,8 +1,8 @@
 (function () {
     const connections = [];
     let connected = false;
+    let couchEventSource;
     const controller = new AbortController();
-    const signal = controller.signal;
 
     self.onconnect = function (e) {
         let port = e.ports[0];
@@ -13,17 +13,19 @@
             connectionId: connections.length
         });
 
-        port.onmessage = async function (event) {
+        port.onmessage = function (event) {
             if (event.data.request === 'close') {
-                console.log('Closing connection');
+                console.debug('🚪 Closing couch connection 🚪');
                 connections.splice(event.data.connectionId - 1, 1);
                 if (connections.length <= 0) {
                     // abort any outstanding requests if there's nobody listening to it.
                     controller.abort();
                 }
 
-                console.log('Closed.');
                 connected = false;
+                // stop listening for events
+                couchEventSource.removeEventListener('message', self.onCouchMessage);
+                console.debug('🚪 Closed couch connection 🚪');
 
                 return;
             }
@@ -33,80 +35,36 @@
                     return;
                 }
 
-                do {
+                self.listenForChanges(event.data.url);
-                    await self.listenForChanges(event.data.url, event.data.body, port);
-                    // eslint-disable-next-line no-unmodified-loop-condition
-                } while (connected);
             }
         };
 
         port.start();
-
     };
 
-    self.onerror = function () {
+    self.onerror = function (error) {
-        //do nothing
+        console.error('🚨 Error on CouchDB feed 🚨', error);
-        console.log('Error on feed');
     };
 
-    self.listenForChanges = async function (url, body, port) {
+    self.onCouchMessage = function (event) {
-        connected = true;
+        console.debug('📩 Received message from CouchDB 📩');
-        let error = false;
+        const objectChanges = JSON.parse(event.data);
-        // feed=continuous maintains an indefinitely open connection with a keep-alive of HEARTBEAT milliseconds until this client closes the connection
+        connections.forEach(function (connection) {
-        // style=main_only returns only the current winning revision of the document
+            connection.postMessage({
-
+                objectChanges
-        console.log('Opening changes feed connection.');
+            });
-        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');
-                console.log('Received chunk');
-                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);
-                        }
-                    });
-                }
-            }
-
-        }
-
-        console.log('Done reading changes feed');
     };
 
+    self.listenForChanges = function (url) {
+        console.debug('⇿ Opening CouchDB change feed connection ⇿');
+
+        couchEventSource = new EventSource(url);
+        couchEventSource.onerror = self.onerror;
+
+        // start listening for events
+        couchEventSource.addEventListener('message', self.onCouchMessage);
+        connected = true;
+        console.debug('⇿ Opened connection ⇿');
+    };
 }());

src/plugins/persistence/couch/CouchObjectProvider.js

@@ -38,6 +38,8 @@ class CouchObjectProvider {
         this.objectQueue = {};
         this.observers = {};
         this.batchIds = [];
+        this.onEventMessage = this.onEventMessage.bind(this);
+        this.onEventError = this.onEventError.bind(this);
     }
 
     /**
@@ -50,7 +52,7 @@ class CouchObjectProvider {
         // eslint-disable-next-line no-undef
         const sharedWorkerURL = `${this.openmct.getAssetPath()}${__OPENMCT_ROOT_RELATIVE__}couchDBChangesFeed.js`;
 
-        sharedWorker = new SharedWorker(sharedWorkerURL);
+        sharedWorker = new SharedWorker(sharedWorkerURL, 'CouchDB SSE Shared Worker');
         sharedWorker.port.onmessage = provider.onSharedWorkerMessage.bind(this);
         sharedWorker.port.onmessageerror = provider.onSharedWorkerMessageError.bind(this);
         sharedWorker.port.start();
@@ -70,6 +72,13 @@ class CouchObjectProvider {
         console.log('Error', event);
     }
 
+    isSynchronizedObject(object) {
+        return (object && object.type
+            && this.openmct.objects.SYNCHRONIZED_OBJECT_TYPES
+            && this.openmct.objects.SYNCHRONIZED_OBJECT_TYPES.includes(object.type));
+
+    }
+
     onSharedWorkerMessage(event) {
         if (event.data.type === 'connection') {
             this.changesFeedSharedWorkerConnectionId = event.data.connectionId;
@@ -86,7 +95,9 @@ class CouchObjectProvider {
             if (observersForObject) {
                 observersForObject.forEach(async (observer) => {
                     const updatedObject = await this.get(objectChanges.identifier);
-                    observer(updatedObject);
+                    if (this.isSynchronizedObject(updatedObject)) {
+                        observer(updatedObject);
+                    }
                 });
             }
         }
@@ -390,38 +401,16 @@ class CouchObjectProvider {
      * @private
      */
     observeObjectChanges() {
-
+        const sseChangesPath = `${this.url}/_changes`;
-        let filter = {selector: {}};
+        const sseURL = new URL(sseChangesPath);
-
+        sseURL.searchParams.append('feed', 'eventsource');
-        if (this.openmct.objects.SYNCHRONIZED_OBJECT_TYPES.length > 1) {
+        sseURL.searchParams.append('style', 'main_only');
-            filter.selector.$or = this.openmct.objects.SYNCHRONIZED_OBJECT_TYPES
+        sseURL.searchParams.append('heartbeat', HEARTBEAT);
-                .map(type => {
-                    return {
-                        'model': {
-                            type
-                        }
-                    };
-                });
-        } else {
-            filter.selector.model = {
-                type: this.openmct.objects.SYNCHRONIZED_OBJECT_TYPES[0]
-            };
-        }
-
-        // 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);
+            this.fetchChanges(sseURL.toString());
         } else {
-            this.initiateSharedWorkerFetchChanges(url, body);
+            this.initiateSharedWorkerFetchChanges(sseURL.toString());
         }
 
     }
@@ -429,7 +418,7 @@ class CouchObjectProvider {
     /**
      * @private
      */
-    initiateSharedWorkerFetchChanges(url, body) {
+    initiateSharedWorkerFetchChanges(url) {
         if (!this.changesFeedSharedWorker) {
             this.changesFeedSharedWorker = this.startSharedWorker();
 
@@ -443,17 +432,40 @@ class CouchObjectProvider {
 
             this.changesFeedSharedWorker.port.postMessage({
                 request: 'changes',
-                body,
                 url
             });
         }
     }
 
-    async fetchChanges(url, body) {
+    onEventError(error) {
-        const controller = new AbortController();
+        console.error('Error on feed', error);
-        const signal = controller.signal;
+        if (Object.keys(this.observers).length > 0) {
+            this.observeObjectChanges();
+        }
+    }
 
-        let error = false;
+    onEventMessage(event) {
+        const object = JSON.parse(event.data);
+        object.identifier = {
+            namespace: this.namespace,
+            key: object.id
+        };
+        let keyString = this.openmct.objects.makeKeyString(object.identifier);
+        let observersForObject = this.observers[keyString];
+
+        if (observersForObject) {
+            observersForObject.forEach(async (observer) => {
+                const updatedObject = await this.get(object.identifier);
+                if (this.isSynchronizedObject(updatedObject)) {
+                    observer(updatedObject);
+                }
+            });
+        }
+    }
+
+    fetchChanges(url) {
+        const controller = new AbortController();
+        let couchEventSource;
 
         if (this.isObservingObjectChanges()) {
             this.stopObservingObjectChanges();
@@ -461,66 +473,18 @@ class CouchObjectProvider {
 
         this.stopObservingObjectChanges = () => {
             controller.abort();
+            couchEventSource.removeEventListener('message', this.onEventMessage);
             delete this.stopObservingObjectChanges;
         };
 
-        const response = await fetch(url, {
+        console.debug('⇿ Opening CouchDB change feed connection ⇿');
-            method: 'POST',
-            signal,
-            headers: {
-                "Content-Type": 'application/json'
-            },
-            body
-        });
 
-        let reader;
+        couchEventSource = new EventSource(url);
+        couchEventSource.onerror = this.onEventError;
 
-        if (response.body === undefined) {
+        // start listening for events
-            error = true;
+        couchEventSource.addEventListener('message', this.onEventMessage);
-        } else {
+        console.debug('⇿ Opened connection ⇿');
-            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 {
-                            const object = JSON.parse(doc);
-                            object.identifier = {
-                                namespace: this.namespace,
-                                key: object.id
-                            };
-                            let keyString = this.openmct.objects.makeKeyString(object.identifier);
-                            let observersForObject = this.observers[keyString];
-
-                            if (observersForObject) {
-                                observersForObject.forEach(async (observer) => {
-                                    const updatedObject = await this.get(object.identifier);
-                                    observer(updatedObject);
-                                });
-                            }
-                        } catch (e) {
-                            //do nothing;
-                        }
-                    });
-                }
-            }
-
-        }
-
-        if (error && Object.keys(this.observers).length > 0) {
-            this.observeObjectChanges();
-        }
     }
 
     /**

src/plugins/persistence/couch/README.md

@@ -30,7 +30,7 @@ openmct.install(openmct.plugins.LocalStorage());
 ```
 Add a line to install the CouchDB plugin for OpenMCT:
 ```
-openmct.install(openmct.plugins.CouchDB("http://localhost:5984/openmct/"));
+openmct.install(openmct.plugins.CouchDB("http://localhost:5984/openmct"));
 ```
 6. Enable cors in CouchDB by editing `/usr/local/etc/local.ini` and add: `
 ```

src/plugins/persistence/couch/pluginSpec.js

@@ -28,7 +28,7 @@ import {
 describe('the plugin', () => {
     let openmct;
     let provider;
-    let testPath = '/test/db';
+    let testPath = 'http://localhost:9990/openmct';
     let options;
     let mockIdentifierService;
     let mockDomainObject;
@@ -66,6 +66,7 @@ describe('the plugin', () => {
         openmct.install(new CouchPlugin(options));
 
         openmct.types.addType('notebook', {creatable: true});
+        openmct.setAssetPath('/base');
 
         openmct.on('start', done);
         openmct.startHeadless();
@@ -74,6 +75,10 @@ describe('the plugin', () => {
         spyOn(provider, 'get').and.callThrough();
         spyOn(provider, 'create').and.callThrough();
         spyOn(provider, 'update').and.callThrough();
+        spyOn(provider, 'startSharedWorker').and.callThrough();
+        spyOn(provider, 'fetchChanges').and.callThrough();
+        spyOn(provider, 'onSharedWorkerMessage').and.callThrough();
+        spyOn(provider, 'onEventMessage').and.callThrough();
     });
 
     afterEach(() => {
@@ -102,12 +107,11 @@ describe('the plugin', () => {
             expect(result.identifier.key).toEqual(mockDomainObject.identifier.key);
         });
 
-        it('creates an object', (done) => {
+        it('creates an object and starts shared worker', async () => {
-            openmct.objects.save(mockDomainObject).then((result) => {
+            const result = await openmct.objects.save(mockDomainObject);
-                expect(provider.create).toHaveBeenCalled();
+            expect(provider.create).toHaveBeenCalled();
-                expect(result).toBeTrue();
+            expect(provider.startSharedWorker).toHaveBeenCalled();
-                done();
+            expect(result).toBeTrue();
-            });
         });
 
         it('updates an object', (done) => {
@@ -123,6 +127,53 @@ describe('the plugin', () => {
                 });
             });
         });
+
+        it('works without Shared Workers', async () => {
+            let sharedWorkerCallback;
+            window.SharedWorker = undefined;
+            const mockEventSource = {
+                addEventListener: (topic, addedListener) => {
+                    sharedWorkerCallback = addedListener;
+                },
+                removeEventListener: () => {
+                    sharedWorkerCallback = null;
+                }
+            };
+            window.EventSource = function (url) {
+                return mockEventSource;
+            };
+
+            mockDomainObject.id = mockDomainObject.identifier.key;
+
+            const fakeUpdateEvent = {
+                data: JSON.stringify(mockDomainObject)
+            };
+
+            // eslint-disable-next-line require-await
+            provider.request = async function (subPath, method, body, signal) {
+                return {
+                    body: fakeUpdateEvent,
+                    ok: true,
+                    id: mockDomainObject.id,
+                    rev: 5
+                };
+            };
+
+            const result = await openmct.objects.save(mockDomainObject);
+            expect(result).toBeTrue();
+            expect(provider.create).toHaveBeenCalled();
+            expect(provider.startSharedWorker).not.toHaveBeenCalled();
+            //Set modified timestamp it detects a change and persists the updated model.
+            mockDomainObject.modified = Date.now();
+            const updatedResult = await openmct.objects.save(mockDomainObject);
+            openmct.objects.observe(mockDomainObject, '*', (updatedObject) => {
+            });
+            expect(updatedResult).toBeTrue();
+            expect(provider.update).toHaveBeenCalled();
+            expect(provider.fetchChanges).toHaveBeenCalled();
+            sharedWorkerCallback(fakeUpdateEvent);
+            expect(provider.onEventMessage).toHaveBeenCalled();
+        });
     });
     describe('batches requests', () => {
         let mockPromise;
@@ -214,6 +265,5 @@ describe('the plugin', () => {
             expect(requestPayload).toBeDefined();
             expect(requestPayload.selector.model.name.$regex).toEqual('(?i)test');
         });
-
     });
 });