ui: Add data-source component and related services (#6486)

* ui: Add data-source component and related services:

1. DataSource component
2. Repository manager for retrieving repositories based on URIs
3. Blocking data service for injection to the data-source component to
support blocking query types of data sources
4. 'Once' promise based data service for injection for potential
fallback to old style promise based data (would need to be injected via
an initial runtime variable)
5. Several utility functions taken from elsewhere
  - maybeCall - a replication of code from elsewhere for condition
  calling a function based on the result of a promise
  - restartWhenAvailable - used for restarting blocking queries when a
  tab is brought to the front
  - ifNotBlocking - to check if blocking is NOT enabled

ui-v2/app/components/data-source.js

@@ -0,0 +1,121 @@
+import Component from '@ember/component';
+import { inject as service } from '@ember/service';
+import { set } from '@ember/object';
+
+import WithListeners from 'consul-ui/mixins/with-listeners';
+
+// Utility function to set, but actually replace if we should replace
+// then call a function on the thing to be replaced (usually a clean up function)
+const replace = function(
+  obj,
+  prop,
+  value,
+  destroy = (prev = null, value) => (typeof prev === 'function' ? prev() : null)
+) {
+  const prev = obj[prop];
+  if (prev !== value) {
+    destroy(prev, value);
+  }
+  return set(obj, prop, value);
+};
+
+/**
+ * @module DataSource
+ *
+ * The DataSource component manages opening and closing data sources via an injectable data service.
+ * Data sources are only opened only if the component is visible in the viewport (using IntersectionObserver).
+ *
+ * Sources returned by the data service should follow an EventTarget/EventSource API.
+ * Management of the caching/usage/counting etc of sources should be done in the data service,
+ * not the component.
+ *
+ * @example ```js
+ *   {{data-source
+ *      src="/dc-1/services/*"
+ *      onchange={{action (mut items) value='data'}}
+ *      onerror={{action (mut error) value='error'}}
+ *   /}}```
+ *
+ * @param src {String} - An identitier used to determine the source of the data. This is passed
+ *                       to the data service for it to determine how to fetch the data.
+ * @param onchange=null {Func} - An action called when the data changes.
+ * @param onerror=null {Func} - An action called on error
+ *
+ */
+export default Component.extend(WithListeners, {
+  tagName: 'span',
+
+  // TODO: can be injected with a simpler non-blocking
+  // data service if we turn off blocking queries completely at runtime
+  data: service('blocking'),
+
+  onchange: function() {},
+  onerror: function() {},
+
+  didInsertElement: function() {
+    this._super(...arguments);
+    const options = {
+      rootMargin: '0px',
+      threshold: 1.0,
+    };
+
+    const observer = new IntersectionObserver((entries, observer) => {
+      entries.map(item => {
+        set(this, 'isIntersecting', item.isIntersecting);
+        if (!item.isIntersecting) {
+          this.actions.close.bind(this)();
+        } else {
+          this.actions.open.bind(this)();
+        }
+      });
+    }, options);
+    observer.observe(this.element); // eslint-disable-line ember/no-observers
+    this.listen(() => {
+      this.actions.close.bind(this)();
+      observer.disconnect(); // eslint-disable-line ember/no-observers
+    });
+  },
+  didReceiveAttrs: function() {
+    this._super(...arguments);
+    if (this.element && this.isIntersecting) {
+      this.actions.open.bind(this)();
+    }
+  },
+  actions: {
+    // keep this argumentless
+    open: function() {
+      const src = this.src;
+      const filter = this.filter;
+
+      // get a new source and replace the old one, cleaning up as we go
+      const source = replace(
+        this,
+        'source',
+        this.data.open(`${src}${filter ? `?filter=${filter}` : ``}`, this),
+        (prev, source) => {
+          // Makes sure any previous source (if different) is ALWAYS closed
+          this.data.close(prev, this);
+        }
+      );
+      // set up the listeners (which auto cleanup on component destruction)
+      const remove = this.listen(source, {
+        message: e => this.onchange(e),
+        error: e => this.onerror(e),
+      });
+      replace(this, '_remove', remove);
+      // dispatch the current data of the source if we have any
+      if (typeof source.getCurrentEvent === 'function') {
+        const currentEvent = source.getCurrentEvent();
+        if (currentEvent) {
+          this.onchange(currentEvent);
+        }
+      }
+    },
+    // keep this argumentless
+    close: function() {
+      this.data.close(this.source, this);
+      replace(this, '_remove', null);
+      set(this, 'source', undefined);
+    },
+  },
+});

ui-v2/app/services/blocking.js

@@ -0,0 +1,91 @@
+import Service, { inject as service } from '@ember/service';
+
+import MultiMap from 'mnemonist/multi-map';
+
+import { BlockingEventSource as EventSource } from 'consul-ui/utils/dom/event-source';
+// utility functions to make the below code easier to read:
+import { ifNotBlocking } from 'consul-ui/services/settings';
+import { restartWhenAvailable } from 'consul-ui/services/client/http';
+import maybeCall from 'consul-ui/utils/maybe-call';
+
+// TODO: Expose sizes of things via env vars
+
+// caches cursors and previous events when the EventSources are destroyed
+const cache = new Map();
+// keeps a record of currently in use EventSources
+const sources = new Map();
+// keeps a count of currently in use EventSources
+const usage = new MultiMap(Set);
+
+export default Service.extend({
+  client: service('client/http'),
+  settings: service('settings'),
+
+  finder: service('repository/manager'),
+
+  open: function(uri, ref) {
+    let source;
+    // Check the cache for an EventSource that is already being used
+    // for this uri. If we don't have one, set one up.
+    if (!sources.has(uri)) {
+      // Setting up involves finding the correct repository method to call
+      // based on the uri, we use the eventually injectable finder for this.
+      const repo = this.finder.fromURI(...uri.split('?filter='));
+      // We then check the to see if this we have previously cached information
+      // for the URI. This data comes from caching this data when the EventSource
+      // is closed and destroyed. We recreate the EventSource from the data from the cache
+      // if so. The data is currently the cursor position and the last received data.
+      let configuration = {};
+      if (cache.has(uri)) {
+        configuration = cache.get(uri);
+      }
+      // tag on the custom createEvent for ember-data
+      configuration.createEvent = repo.reconcile;
+
+      // We create the EventSource, checking to make sure whether we should close the
+      // EventSource on first response (depending on the user setting) and reopening
+      // the EventSource if it has been paused by the user navigating to another tab
+      source = new EventSource((configuration, source) => {
+        const close = source.close.bind(source);
+        const deleteCursor = () => (configuration.cursor = undefined);
+        //
+        return maybeCall(deleteCursor, ifNotBlocking(this.settings))().then(() => {
+          return repo
+            .find(configuration)
+            .then(maybeCall(close, ifNotBlocking(this.settings)))
+            .catch(restartWhenAvailable(this.client));
+        });
+      }, configuration);
+      // Lastly, when the EventSource is no longer needed, cache its information
+      // for when we next need it again (see above re: data cache)
+      source.addEventListener('close', function close(e) {
+        const source = e.target;
+        source.removeEventListener('close', close);
+        cache.set(uri, {
+          currentEvent: e.target.getCurrentEvent(),
+          cursor: e.target.configuration.cursor,
+        });
+        // the data is cached delete the EventSource
+        sources.delete(uri);
+      });
+      sources.set(uri, source);
+    } else {
+      source = sources.get(uri);
+    }
+    // set/increase the usage counter
+    usage.set(source, ref);
+    source.open();
+    return source;
+  },
+  close: function(source, ref) {
+    if (source) {
+      // decrease the usage counter
+      usage.remove(source, ref);
+      // if the EventSource is no longer being used
+      // close it (data caching is dealt with by the above 'close' event listener)
+      if (!usage.has(source)) {
+        source.close();
+      }
+    }
+  },
+});

ui-v2/app/services/client/http.js

@@ -7,6 +7,21 @@ import getObjectPool from 'consul-ui/utils/get-object-pool';
 import Request from 'consul-ui/utils/http/request';
 import createURL from 'consul-ui/utils/createURL';
 
+// reopen EventSources if a user changes tab
+export const restartWhenAvailable = function(client) {
+  return function(e) {
+    // setup the aborted connection restarting
+    // this should happen here to avoid cache deletion
+    const status = get(e, 'errors.firstObject.status');
+    if (status === '0') {
+      // Any '0' errors (abort) should possibly try again, depending upon the circumstances
+      // whenAvailable returns a Promise that resolves when the client is available
+      // again
+      return client.whenAvailable(e);
+    }
+    throw e;
+  };
+};
 class HTTPError extends Error {
   constructor(statusCode, message) {
     super(message);

ui-v2/app/services/promised.js

@@ -0,0 +1,23 @@
+import Service, { inject as service } from '@ember/service';
+
+import { CallableEventSource as EventSource } from 'consul-ui/utils/dom/event-source';
+
+export default Service.extend({
+  finder: service('repository/manager'),
+
+  open: function(uri, ref) {
+    const repo = this.finder.fromURI(...uri.split('?filter='));
+    const source = new EventSource(function(configuration, source) {
+      return repo.find(configuration).then(function(res) {
+        source.dispatchEvent({ type: 'message', data: res });
+        source.close();
+      });
+    }, {});
+    return source;
+  },
+  close: function(source, ref) {
+    if (source) {
+      source.close();
+    }
+  },
+});

ui-v2/app/services/repository.js

@@ -13,6 +13,18 @@ export default Service.extend({
   },
   //
   store: service('store'),
+  reconcile: function(meta = {}) {
+    // unload anything older than our current sync date/time
+    // FIXME: This needs fixing once again to take nspaces into account
+    if (typeof meta.date !== 'undefined') {
+      this.store.peekAll(this.getModelName()).forEach(item => {
+        const date = item.SyncTime;
+        if (typeof date !== 'undefined' && date != meta.date) {
+          this.store.unloadRecord(item);
+        }
+      });
+    }
+  },
   findAllByDatacenter: function(dc, nspace, configuration = {}) {
     const query = {
       dc: dc,

ui-v2/app/services/repository/manager.js

@@ -0,0 +1,80 @@
+import Service, { inject as service } from '@ember/service';
+
+export default Service.extend({
+  // TODO: Temporary repo list here
+  service: service('repository/service'),
+  node: service('repository/node'),
+  session: service('repository/session'),
+  proxy: service('repository/proxy'),
+
+  fromURI: function(src, filter) {
+    let temp = src.split('/');
+    temp.shift();
+    const dc = temp.shift();
+    const model = temp.shift();
+    const repo = this[model];
+    let slug = temp.join('/');
+
+    // TODO: if something is filtered we shouldn't reconcile things
+    // custom createEvent, here used to reconcile the ember-data store for each tick
+    // ideally we wouldn't do this here, but we handily have access to the repo here
+    const obj = {
+      reconcile: function(result = {}, configuration) {
+        const event = {
+          type: 'message',
+          data: result,
+        };
+        // const meta = get(event.data || {}, 'meta') || {};
+        repo.reconcile(event.data.meta);
+        return event;
+      },
+    };
+    let id, node;
+    switch (slug) {
+      case '*':
+        switch (model) {
+          default:
+            obj.find = function(configuration) {
+              return repo.findAllByDatacenter(dc, configuration);
+            };
+        }
+        break;
+      default:
+        switch (model) {
+          case 'session':
+            obj.find = function(configuration) {
+              return repo.findByNode(slug, dc, configuration);
+            };
+            break;
+          case 'service-instance':
+            temp = slug.split('/');
+            id = temp[0];
+            node = temp[1];
+            slug = temp[2];
+            obj.find = function(configuration) {
+              return repo.findInstanceBySlug(id, node, slug, dc, configuration);
+            };
+            break;
+          case 'service':
+            obj.find = function(configuration) {
+              return repo.findBySlug(slug, dc, configuration);
+            };
+            break;
+          case 'proxy':
+            temp = slug.split('/');
+            id = temp[0];
+            node = temp[1];
+            slug = temp[2];
+            obj.find = function(configuration) {
+              return repo.findInstanceBySlug(id, node, slug, dc, configuration);
+            };
+            break;
+          default:
+            obj.find = function(configuration) {
+              return repo.findBySlug(slug, dc, configuration);
+            };
+        }
+    }
+    return obj;
+  },
+});

ui-v2/app/services/settings.js

@@ -3,6 +3,12 @@ import { Promise } from 'rsvp';
 import getStorage from 'consul-ui/utils/storage/local-storage';
 const SCHEME = 'consul';
 const storage = getStorage(SCHEME);
+// promise aware assertion
+export const ifNotBlocking = function(repo) {
+  return repo.findBySlug('client').then(function(settings) {
+    return typeof settings.blocking !== 'undefined' && !settings.blocking;
+  });
+};
 export default Service.extend({
   storage: storage,
   findHeaders: function() {

ui-v2/app/utils/maybe-call.js

@@ -0,0 +1,17 @@
+/**
+ * Promise aware conditional function call
+ *
+ * @param {function} cb - The function to possibily call
+ * @param {function} [what] - A function returning a boolean resolving promise
+ * @returns {function} - function when called returns a Promise that resolves the argument it is called with
+ */
+export default function(cb, what) {
+  return function(res) {
+    return what.then(function(bool) {
+      if (bool) {
+        cb();
+      }
+      return res;
+    });
+  };
+}

ui-v2/package.json

@@ -104,6 +104,7 @@
     "lint-staged": "^9.2.5",
     "loader.js": "^4.7.0",
     "ngraph.graph": "^18.0.3",
+    "mnemonist": "^0.30.0",
     "node-sass": "^4.9.3",
     "prettier": "^1.10.2",
     "qunit-dom": "^0.9.0",