Auto refresh on conflict

src/api/objects/ConflictError.js

@@ -0,0 +1,2 @@
+export default class ConflictError extends Error {
+}

src/api/objects/ObjectAPI.js

@@ -26,6 +26,7 @@ import RootRegistry from './RootRegistry';
 import RootObjectProvider from './RootObjectProvider';
 import EventEmitter from 'EventEmitter';
 import InterceptorRegistry from './InterceptorRegistry';
+import ConflictError from './ConflictError';
 
 /**
  * Utilities for loading, saving, and manipulating domain objects.
@@ -34,6 +35,7 @@ import InterceptorRegistry from './InterceptorRegistry';
  */
 
 function ObjectAPI(typeRegistry, openmct) {
+    this.openmct = openmct;
     this.typeRegistry = typeRegistry;
     this.eventEmitter = new EventEmitter();
     this.providers = {};
@@ -47,6 +49,10 @@ function ObjectAPI(typeRegistry, openmct) {
     this.interceptorRegistry = new InterceptorRegistry();
 
     this.SYNCHRONIZED_OBJECT_TYPES = ['notebook', 'plan'];
+
+    this.errors = {
+        Conflict: ConflictError
+    };
 }
 
 /**
@@ -298,10 +304,11 @@ ObjectAPI.prototype.save = function (domainObject) {
                 savedResolve = resolve;
             });
             domainObject.persisted = persistedTime;
-            provider.create(domainObject).then((response) => {
+            provider.create(domainObject)
-                this.mutate(domainObject, 'persisted', persistedTime);
+                .then((response) => {
-                savedResolve(response);
+                    this.mutate(domainObject, 'persisted', persistedTime);
-            });
+                    savedResolve(response);
+                });
         } else {
             domainObject.persisted = persistedTime;
             this.mutate(domainObject, 'persisted', persistedTime);
@@ -309,7 +316,23 @@ ObjectAPI.prototype.save = function (domainObject) {
         }
     }
 
-    return result;
+    return result.catch((error) => {
+        if (error instanceof this.errors.Conflict) {
+            if (this.supportsMutation(domainObject.identifier)) {
+                return this.getMutable(domainObject.identifier).then((mutable) => {
+                    mutable.$refresh(mutable);
+                    this.destroyMutable(mutable);
+                    this.openmct.notifications.alert(`Conflict while saving ${domainObject.name}. Please try again.`, {
+                        autoDismissTimeout: 5000
+                    });
+
+                    return true;
+                });
+            }
+        }
+
+        throw error;
+    });
 };
 
 /**

src/installDefaultBundles.js

@@ -41,7 +41,6 @@ const DEFAULTS = [
     'platform/forms',
     'platform/identity',
     'platform/persistence/aggregator',
-    'platform/persistence/queue',
     'platform/policy',
     'platform/entanglement',
     'platform/search',

src/plugins/persistence/couch/CouchChangesFeed.js

@@ -15,12 +15,16 @@
 
         port.onmessage = async function (event) {
             if (event.data.request === 'close') {
+                console.log('Closing 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;
+
                 return;
             }
 
@@ -29,68 +33,9 @@
                     return;
                 }
 
-                connected = true;
+                do {
-
+                    await self.listenForChanges(event.data.url, event.data.body, port);
-                let url = event.data.url;
+                } while (connected);
-                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
-                    });
-                }
             }
         };
 
@@ -103,4 +48,71 @@
         console.log('Error on feed');
     };
 
+    self.listenForChanges = async function (url, body, port) {
+        connected = true;
+        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
+
+        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');
+
+        if (error) {
+            console.log(`Finished with error ${error.toString()}`);
+            port.postMessage({
+                error
+            });
+        }
+    };
+
 }());

src/plugins/persistence/couch/CouchObjectProvider.js

@@ -126,11 +126,12 @@ export default class CouchObjectProvider {
         }
 
         return fetch(this.url + '/' + subPath, fetchOptions)
-            .then(response => response.json())
+            .then((response) => {
-            .then(function (response) {
+                if (response.status === 409) {
-                return response;
+                    throw new this.openmct.objects.errors.Conflict(`Conflict persisting ${fetchOptions.body.name}`);
-            }, function () {
+                }
-                return undefined;
+
+                return response.json();
             });
     }
 
@@ -581,6 +582,9 @@ export default class CouchObjectProvider {
             let document = new CouchDocument(key, queued.model, this.objectQueue[key].rev);
             this.request(key, "PUT", document).then((response) => {
                 this.checkResponse(response, queued.intermediateResponse, key);
+            }).catch((error) => {
+                queued.intermediateResponse.reject(error);
+                this.objectQueue[key].pending = false;
             });
         }
     }