From 0a47a95588a5dd4ae76ad088665fcb1720da51d5 Mon Sep 17 00:00:00 2001
From: John Cowen <johncowen@users.noreply.github.com>
Date: Thu, 19 Mar 2020 10:28:21 +0000
Subject: [PATCH] ui: Add DataSource component (#7448)

* 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

* Move to a different organization based on protocols

* Don't call open twice when eager

* Workaround new ember error for reading and writing at the same time

* Add first draft of a README.mdx file
---
 ui-v2/app/components/data-source/README.mdx   |  61 +++++++
 ui-v2/app/components/data-source/index.hbs    |   4 +
 ui-v2/app/components/data-source/index.js     | 150 ++++++++++++++++++
 .../services/data-source/protocols/http.js    |  34 ++++
 .../data-source/protocols/http/blocking.js    |  30 ++++
 .../data-source/protocols/http/promise.js     |   8 +
 .../data-source/protocols/local-storage.js    |  17 ++
 ui-v2/app/services/data-source/service.js     |  84 ++++++++++
 ui-v2/app/services/repository.js              |  12 ++
 ui-v2/app/services/settings.js                |   6 +
 ui-v2/app/utils/dom/event-source/index.js     |  13 ++
 ui-v2/app/utils/dom/event-source/storage.js   |  38 +++--
 ui-v2/app/utils/maybe-call.js                 |  17 ++
 ui-v2/app/utils/storage/local-storage.js      |  13 +-
 ui-v2/package.json                            |   1 +
 .../components/data-source-test.js            | 121 ++++++++++++++
 ui-v2/tests/steps.js                          |  37 +++--
 ui-v2/tests/steps/assertions/dom.js           |   9 +-
 ui-v2/tests/steps/assertions/model.js         |   8 +-
 .../data-source/protocols/http-test.js        |  12 ++
 .../protocols/local-storage-test.js           |  12 ++
 ui-v2/tests/unit/utils/maybe-call-test.js     |  18 +++
 ui-v2/yarn.lock                               |  25 ++-
 23 files changed, 691 insertions(+), 39 deletions(-)
 create mode 100644 ui-v2/app/components/data-source/README.mdx
 create mode 100644 ui-v2/app/components/data-source/index.hbs
 create mode 100644 ui-v2/app/components/data-source/index.js
 create mode 100644 ui-v2/app/services/data-source/protocols/http.js
 create mode 100644 ui-v2/app/services/data-source/protocols/http/blocking.js
 create mode 100644 ui-v2/app/services/data-source/protocols/http/promise.js
 create mode 100644 ui-v2/app/services/data-source/protocols/local-storage.js
 create mode 100644 ui-v2/app/services/data-source/service.js
 create mode 100644 ui-v2/app/utils/maybe-call.js
 create mode 100644 ui-v2/tests/integration/components/data-source-test.js
 create mode 100644 ui-v2/tests/unit/services/data-source/protocols/http-test.js
 create mode 100644 ui-v2/tests/unit/services/data-source/protocols/local-storage-test.js
 create mode 100644 ui-v2/tests/unit/utils/maybe-call-test.js

diff --git a/ui-v2/app/components/data-source/README.mdx b/ui-v2/app/components/data-source/README.mdx
new file mode 100644
index 000000000..63c485a3d
--- /dev/null
+++ b/ui-v2/app/components/data-source/README.mdx
@@ -0,0 +1,61 @@
+## DataSource
+
+```handlebars
+<DataSource
+  @src="/dc/nspace/services"
+  @loading="eager"
+  @onchange={{action (mut items) value="data"}}
+  @onerror={{action (mut error) value="error"}}
+/>
+```
+
+### Arguments
+
+| Argument | Type | Default | Description |
+| --- | --- | --- | --- |
+| `src` | `String` | | The source to subscribe to updates to, this should map to a string based URI |
+| `loading` | `String` | eager | Allows the browser to defer loading offscreen DataSources (`eager\|lazy`). Setting to `lazy` only loads the data when the DataSource is visible in the DOM (inc. `display: none\|block;`) |
+| `onchange` | `Function` |  | The action to fire when the data changes. Emits an Event-like object with a `data` property containing the data. |
+| `onerror` | `Function` |  | The action to fire when an error occurs. Emits ErrorEvent object with an `error` property containing the Error. |
+
+The component takes a `src` or an identifier (a uri) for some data and then emits `onchange` events whenever that data changes. If an error occurs whilst listening for data changes, an `onerror` event is emitted.
+
+Setting `@loading="lazy"` uses `IntersectionObserver` to activate/deactive event emitting until the `<DataSource />` element is displayed in the DOM. This means you can use CSS `display: none|block;` to control the loading and stopping loading of data for usage with CSS based tabs and such-like.
+
+Consuls HTTP API DataSources use Consul's blocking query support for live updating of data.
+
+Behind the scenes in the Consul UI we map URIs back to our `ember-data` backed `Repositories` meaning we can essentially redesign the URIs used for our data to more closely fit our needs. For example we currently require that **all** HTTP API URIs begin with `/dc/nspace/` values whether they require them or not.
+
+`DataSource` is not just restricted to HTTP API data, and can be configured to listen for data changes using a variety of methods and sources. For example we have also configured `DataSource` to listen to `LocalStorage` changes using the `settings://` pseudo-protocol in the URI (See examples below).
+
+
+### Example
+
+Straightforward usage can use `mut` to easily update data within a template
+
+```handlebars
+  {{! listen for HTTP API changes}}
+  <DataSource @src="/dc/nspace/services"
+    @onchange={{action (mut items) value="data"}}
+    @onerror={{action (mut error) value="error"}}
+  />
+  {{! the value of items will change whenever the data changes}}
+  {{#each items as |item|}}
+    {{item.Name}} {{! < Prints the item name }}
+  {{/each}}
+
+  {{! listen for Settings (local storage) changes}}
+  <DataSource @src="settings://consul:token"
+    @onchange={{action (mut token) value="data"}}
+    @onerror={{action (mut error) value="error"}}
+  />
+  {{! the value of token will change whenever the data changes}}
+  {{token.AccessorID}} {{! < Prints the token AccessorID }}
+```
+
+### See
+
+- [Component Source Code](./index.js)
+- [Template Source Code](./index.hbs)
+
+---
diff --git a/ui-v2/app/components/data-source/index.hbs b/ui-v2/app/components/data-source/index.hbs
new file mode 100644
index 000000000..661f27adb
--- /dev/null
+++ b/ui-v2/app/components/data-source/index.hbs
@@ -0,0 +1,4 @@
+{{#if (eq loading "lazy")}}
+{{! in order to use intersection observer we need a DOM element on the page}}
+  <data aria-hidden="true" style="width: 0;height: 0;font-size: 0;padding: 0;margin: 0;" />
+{{/if}}
diff --git a/ui-v2/app/components/data-source/index.js b/ui-v2/app/components/data-source/index.js
new file mode 100644
index 000000000..dd41b0ce4
--- /dev/null
+++ b/ui-v2/app/components/data-source/index.js
@@ -0,0 +1,150 @@
+import Component from '@ember/component';
+import { inject as service } from '@ember/service';
+import { set } from '@ember/object';
+
+import Ember from 'ember';
+/**
+ * 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)
+ *
+ * @param obj - target object with the property to replace
+ * @param prop {string} - property to replace on the target object
+ * @param value - value to use for replacement
+ * @param destroy {(prev: any, value: any) => any} - teardown 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 ```javascript
+ *   <DataSource
+ *      src="/dc-1/~nspace/services"
+ *      onchange={{action (mut items) value='data'}}
+ *      onerror={{action (mut error) value='error'}}
+ *   />```
+ *
+ * @param src {string} - An identifier used to determine the source of the data. This is passed
+ * @param loading {string} - Either `eager` or `lazy`, lazy will only load the data once the component
+ *                           is in the viewport
+ * @param onchange {function=} - An action called when the data changes.
+ * @param onerror {function=} - An action called on error
+ *
+ */
+export default Component.extend({
+  tagName: '',
+
+  data: service('data-source/service'),
+  dom: service('dom'),
+  logger: service('logger'),
+
+  onchange: function(e) {},
+  onerror: function(e) {},
+
+  loading: 'eager',
+
+  isIntersecting: false,
+
+  init: function() {
+    this._super(...arguments);
+    this._listeners = this.dom.listeners();
+    this._lazyListeners = this.dom.listeners();
+  },
+  willDestroy: function() {
+    this.actions.close.apply(this);
+    this._listeners.remove();
+    this._lazyListeners.remove();
+  },
+
+  didInsertElement: function() {
+    this._super(...arguments);
+    if (this.loading === 'lazy') {
+      this._lazyListeners.add(
+        this.dom.isInViewport(this.element, inViewport => {
+          set(this, 'isIntersecting', inViewport || Ember.testing);
+          if (!this.isIntersecting) {
+            this.actions.close.bind(this)();
+          } else {
+            this.actions.open.bind(this)();
+          }
+        })
+      );
+    }
+  },
+  didReceiveAttrs: function() {
+    this._super(...arguments);
+    if (this.loading === 'eager') {
+      this._lazyListeners.remove();
+    }
+    if (this.loading === 'eager' || this.isIntersecting) {
+      this.actions.open.bind(this)();
+    }
+  },
+  actions: {
+    // keep this argumentless
+    open: function() {
+      // get a new source and replace the old one, cleaning up as we go
+      const source = replace(this, 'source', this.data.open(this.src, this), (prev, source) => {
+        // Makes sure any previous source (if different) is ALWAYS closed
+        this.data.close(prev, this);
+      });
+      const error = err => {
+        try {
+          this.onerror(err);
+          this.logger.execute(err);
+        } catch (err) {
+          this.logger.execute(err);
+        }
+      };
+      // set up the listeners (which auto cleanup on component destruction)
+      const remove = this._listeners.add(this.source, {
+        message: e => {
+          try {
+            this.onchange(e);
+          } catch (err) {
+            error(err);
+          }
+        },
+        error: e => error(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) {
+          try {
+            this.onchange(currentEvent);
+          } catch (err) {
+            error(err);
+          }
+        }
+      }
+    },
+    // keep this argumentless
+    close: function() {
+      if (typeof this.source !== 'undefined') {
+        this.data.close(this.source, this);
+        replace(this, '_remove', undefined);
+        set(this, 'source', undefined);
+      }
+    },
+  },
+});
diff --git a/ui-v2/app/services/data-source/protocols/http.js b/ui-v2/app/services/data-source/protocols/http.js
new file mode 100644
index 000000000..3440f8666
--- /dev/null
+++ b/ui-v2/app/services/data-source/protocols/http.js
@@ -0,0 +1,34 @@
+import Service, { inject as service } from '@ember/service';
+import { get } from '@ember/object';
+
+export default Service.extend({
+  datacenters: service('repository/dc'),
+  token: service('repository/token'),
+  type: service('data-source/protocols/http/blocking'),
+  source: function(src, configuration) {
+    const [, dc /*nspace*/, , model, ...rest] = src.split('/');
+    let find;
+    const repo = this[model];
+    if (typeof repo.reconcile === 'function') {
+      configuration.createEvent = function(result = {}, configuration) {
+        const event = {
+          type: 'message',
+          data: result,
+        };
+        if (repo.reconcile === 'function') {
+          repo.reconcile(get(event, 'data.meta') || {});
+        }
+        return event;
+      };
+    }
+    switch (model) {
+      case 'datacenters':
+        find = configuration => repo.findAll(configuration);
+        break;
+      case 'token':
+        find = configuration => repo.self(rest[1], dc);
+        break;
+    }
+    return this.type.source(find, configuration);
+  },
+});
diff --git a/ui-v2/app/services/data-source/protocols/http/blocking.js b/ui-v2/app/services/data-source/protocols/http/blocking.js
new file mode 100644
index 000000000..9e363ea1d
--- /dev/null
+++ b/ui-v2/app/services/data-source/protocols/http/blocking.js
@@ -0,0 +1,30 @@
+import Service, { inject as service } from '@ember/service';
+import { get } from '@ember/object';
+
+import { BlockingEventSource as EventSource } from 'consul-ui/utils/dom/event-source';
+import { ifNotBlocking } from 'consul-ui/services/settings';
+import { restartWhenAvailable } from 'consul-ui/services/client/http';
+import maybeCall from 'consul-ui/utils/maybe-call';
+
+export default Service.extend({
+  client: service('client/http'),
+  settings: service('settings'),
+  source: function(find, configuration) {
+    return new EventSource((configuration, source) => {
+      const close = source.close.bind(source);
+      const deleteCursor = () => (configuration.cursor = undefined);
+      //
+      return maybeCall(deleteCursor, ifNotBlocking(this.settings))().then(() => {
+        return find(configuration)
+          .then(maybeCall(close, ifNotBlocking(this.settings)))
+          .then(function(res) {
+            if (typeof get(res, 'meta.cursor') === 'undefined') {
+              close();
+            }
+            return res;
+          })
+          .catch(restartWhenAvailable(this.client));
+      });
+    }, configuration);
+  },
+});
diff --git a/ui-v2/app/services/data-source/protocols/http/promise.js b/ui-v2/app/services/data-source/protocols/http/promise.js
new file mode 100644
index 000000000..cc1d55c00
--- /dev/null
+++ b/ui-v2/app/services/data-source/protocols/http/promise.js
@@ -0,0 +1,8 @@
+import Service from '@ember/service';
+import { once } from 'consul-ui/utils/dom/event-source';
+
+export default Service.extend({
+  source: function(find, configuration) {
+    return once(find, configuration);
+  },
+});
diff --git a/ui-v2/app/services/data-source/protocols/local-storage.js b/ui-v2/app/services/data-source/protocols/local-storage.js
new file mode 100644
index 000000000..846d68459
--- /dev/null
+++ b/ui-v2/app/services/data-source/protocols/local-storage.js
@@ -0,0 +1,17 @@
+import Service, { inject as service } from '@ember/service';
+import { StorageEventSource } from 'consul-ui/utils/dom/event-source';
+
+export default Service.extend({
+  repo: service('settings'),
+  source: function(src, configuration) {
+    const slug = src.split(':').pop();
+    return new StorageEventSource(
+      configuration => {
+        return this.repo.findBySlug(slug);
+      },
+      {
+        key: src,
+      }
+    );
+  },
+});
diff --git a/ui-v2/app/services/data-source/service.js b/ui-v2/app/services/data-source/service.js
new file mode 100644
index 000000000..9f8f14c75
--- /dev/null
+++ b/ui-v2/app/services/data-source/service.js
@@ -0,0 +1,84 @@
+import Service, { inject as service } from '@ember/service';
+
+import MultiMap from 'mnemonist/multi-map';
+
+// TODO: Expose sizes of things via env vars
+
+// caches cursors and previous events when the EventSources are destroyed
+let cache;
+// keeps a record of currently in use EventSources
+let sources;
+// keeps a count of currently in use EventSources
+let usage;
+
+export default Service.extend({
+  dom: service('dom'),
+  consul: service('data-source/protocols/http'),
+  settings: service('data-source/protocols/local-storage'),
+
+  init: function() {
+    this._super(...arguments);
+    cache = new Map();
+    sources = new Map();
+    usage = new MultiMap(Set);
+    this._listeners = this.dom.listeners();
+  },
+  willDestroy: function() {
+    this._listeners.remove();
+  },
+
+  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 (uri.indexOf('://') === -1) {
+      uri = `consul://${uri}`;
+    }
+    if (!sources.has(uri)) {
+      const url = new URL(uri);
+      let pathname = url.pathname;
+      if (pathname.startsWith('//')) {
+        pathname = pathname.substr(2);
+      }
+      const providerName = url.protocol.substr(0, url.protocol.length - 1);
+
+      const provider = this[providerName];
+
+      let configuration = {};
+      if (cache.has(uri)) {
+        configuration = cache.get(uri);
+      }
+      source = provider.source(pathname, configuration);
+      this._listeners.add(source, {
+        close: e => {
+          const source = e.target;
+          source.removeEventListener('close', close);
+          cache.set(uri, {
+            currentEvent: source.getCurrentEvent(),
+            cursor: source.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();
+      }
+    }
+  },
+});
diff --git a/ui-v2/app/services/repository.js b/ui-v2/app/services/repository.js
index 7cbc88e91..202378268 100644
--- a/ui-v2/app/services/repository.js
+++ b/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,
diff --git a/ui-v2/app/services/settings.js b/ui-v2/app/services/settings.js
index 8262400fe..7870a8ac1 100644
--- a/ui-v2/app/services/settings.js
+++ b/ui-v2/app/services/settings.js
@@ -2,6 +2,12 @@ import Service from '@ember/service';
 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,
   findAll: function(key) {
diff --git a/ui-v2/app/utils/dom/event-source/index.js b/ui-v2/app/utils/dom/event-source/index.js
index f29854a56..06d6e3182 100644
--- a/ui-v2/app/utils/dom/event-source/index.js
+++ b/ui-v2/app/utils/dom/event-source/index.js
@@ -131,3 +131,16 @@ export const toPromise = function(target, cb, eventName = 'message', errorName =
     cb(remove);
   });
 };
+export const once = function(cb, configuration, Source = OpenableEventSource) {
+  return new Source(function(configuration, source) {
+    return cb(configuration, source)
+      .then(function(data) {
+        source.dispatchEvent({ type: 'message', data: data });
+        source.close();
+      })
+      .catch(function(e) {
+        source.dispatchEvent({ type: 'error', error: e });
+        source.close();
+      });
+  }, configuration);
+};
diff --git a/ui-v2/app/utils/dom/event-source/storage.js b/ui-v2/app/utils/dom/event-source/storage.js
index 26f2ab48b..602c6af3d 100644
--- a/ui-v2/app/utils/dom/event-source/storage.js
+++ b/ui-v2/app/utils/dom/event-source/storage.js
@@ -1,20 +1,27 @@
 export default function(EventTarget, P = Promise) {
   const handler = function(e) {
-    if (e.key === this.configuration.key) {
-      P.resolve(this.getCurrentEvent()).then(event => {
-        this.configuration.cursor++;
-        this.dispatchEvent(event);
-      });
+    // e is undefined on the opening call
+    if (typeof e === 'undefined' || e.key === this.configuration.key) {
+      if (this.readyState === 1) {
+        const res = this.source(this.configuration);
+        P.resolve(res).then(data => {
+          this.configuration.cursor++;
+          this._currentEvent = { type: 'message', data: data };
+          this.dispatchEvent({ type: 'message', data: data });
+        });
+      }
     }
   };
   return class extends EventTarget {
     constructor(cb, configuration) {
       super(...arguments);
+      this.readyState = 2;
+      this.target = configuration.target || window;
+      this.name = 'storage';
       this.source = cb;
       this.handler = handler.bind(this);
       this.configuration = configuration;
       this.configuration.cursor = 1;
-      this.dispatcher = configuration.dispatcher;
       this.open();
     }
     dispatchEvent() {
@@ -23,18 +30,19 @@ export default function(EventTarget, P = Promise) {
       }
     }
     close() {
-      this.dispatcher.removeEventListener('storage', this.handler);
+      this.target.removeEventListener(this.name, this.handler);
       this.readyState = 2;
     }
-    reopen() {
-      this.dispatcher.addEventListener('storage', this.handler);
-      this.readyState = 1;
-    }
     getCurrentEvent() {
-      return {
-        type: 'message',
-        data: this.source(this.configuration),
-      };
+      return this._currentEvent;
+    }
+    open() {
+      const state = this.readyState;
+      this.readyState = 1;
+      if (state !== 1) {
+        this.target.addEventListener(this.name, this.handler);
+        this.handler();
+      }
     }
   };
 }
diff --git a/ui-v2/app/utils/maybe-call.js b/ui-v2/app/utils/maybe-call.js
new file mode 100644
index 000000000..33f73bc22
--- /dev/null
+++ b/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;
+    });
+  };
+}
diff --git a/ui-v2/app/utils/storage/local-storage.js b/ui-v2/app/utils/storage/local-storage.js
index 0e17382f0..9f8e1541d 100644
--- a/ui-v2/app/utils/storage/local-storage.js
+++ b/ui-v2/app/utils/storage/local-storage.js
@@ -2,7 +2,10 @@ export default function(
   scheme = '',
   storage = window.localStorage,
   encode = JSON.stringify,
-  decode = JSON.parse
+  decode = JSON.parse,
+  dispatch = function(key) {
+    window.dispatchEvent(new StorageEvent('storage', { key: key }));
+  }
 ) {
   const prefix = `${scheme}:`;
   return {
@@ -27,10 +30,14 @@ export default function(
       } catch (e) {
         value = '""';
       }
-      return storage.setItem(`${prefix}${path}`, value);
+      const res = storage.setItem(`${prefix}${path}`, value);
+      dispatch(`${prefix}${path}`);
+      return res;
     },
     removeValue: function(path) {
-      return storage.removeItem(`${prefix}${path}`);
+      const res = storage.removeItem(`${prefix}${path}`);
+      dispatch(`${prefix}${path}`);
+      return res;
     },
     all: function() {
       return Object.keys(storage).reduce((prev, item, i, arr) => {
diff --git a/ui-v2/package.json b/ui-v2/package.json
index e8b59c298..3eaa98b7f 100644
--- a/ui-v2/package.json
+++ b/ui-v2/package.json
@@ -116,6 +116,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",
     "pretender": "^3.2.0",
     "prettier": "^1.10.2",
diff --git a/ui-v2/tests/integration/components/data-source-test.js b/ui-v2/tests/integration/components/data-source-test.js
new file mode 100644
index 000000000..530dbb0a3
--- /dev/null
+++ b/ui-v2/tests/integration/components/data-source-test.js
@@ -0,0 +1,121 @@
+import { module } from 'qunit';
+import { setupRenderingTest } from 'ember-qunit';
+import { clearRender, render, waitUntil } from '@ember/test-helpers';
+import hbs from 'htmlbars-inline-precompile';
+
+import test from 'ember-sinon-qunit/test-support/test';
+import Service from '@ember/service';
+
+import { BlockingEventSource as RealEventSource } from 'consul-ui/utils/dom/event-source';
+
+const createFakeBlockingEventSource = function() {
+  const EventSource = function(cb) {
+    this.readyState = 1;
+    this.source = cb;
+  };
+  const o = EventSource.prototype;
+  [
+    'addEventListener',
+    'removeEventListener',
+    'dispatchEvent',
+    'close',
+    'open',
+    'getCurrentEvent',
+  ].forEach(function(item) {
+    o[item] = function() {};
+  });
+  return EventSource;
+};
+const BlockingEventSource = createFakeBlockingEventSource();
+module('Integration | Component | data-source', function(hooks) {
+  setupRenderingTest(hooks);
+
+  hooks.beforeEach(function() {
+    this.actions = {};
+    this.send = (actionName, ...args) => this.actions[actionName].apply(this, args);
+  });
+  test('open and closed are called correctly when the src is changed', async function(assert) {
+    // Set any properties with this.set('myProperty', 'value');
+    // Handle any actions with this.set('myAction', function(val) { ... });
+    assert.expect(9);
+    const close = this.stub();
+    const open = this.stub();
+    const addEventListener = this.stub();
+    const removeEventListener = this.stub();
+    let count = 0;
+    const fakeService = Service.extend({
+      open: function(uri, obj) {
+        open(uri);
+        const source = new BlockingEventSource();
+        source.getCurrentEvent = function() {
+          return { data: uri };
+        };
+        source.addEventListener = addEventListener;
+        source.removeEventListener = removeEventListener;
+        return source;
+      },
+      close: close,
+    });
+    this.owner.register('service:data-source/fake-service', fakeService);
+    this.owner.inject('component:data-source', 'data', 'service:data-source/fake-service');
+    this.actions.change = data => {
+      count++;
+      switch (count) {
+        case 1:
+          assert.equal(data, 'a', 'change was called first with "a"');
+          setTimeout(() => this.set('src', 'b'), 0);
+          break;
+        case 2:
+          assert.equal(data, 'b', 'change was called second with "b"');
+          break;
+      }
+    };
+
+    this.set('src', 'a');
+    await render(hbs`<DataSource @src={{src}} @onchange={{action "change" value="data"}} />`);
+    assert.equal(this.element.textContent.trim(), '');
+    await waitUntil(() => {
+      return close.calledTwice;
+    });
+    assert.ok(open.calledTwice, 'open is called when src is set');
+    assert.ok(close.calledTwice, 'close is called as open is called');
+    await clearRender();
+    await waitUntil(() => {
+      return close.calledThrice;
+    });
+    assert.ok(open.calledTwice, 'open is _still_ only called when src is set');
+    assert.ok(close.calledThrice, 'close is called an extra time as the component is destroyed');
+    assert.equal(addEventListener.callCount, 4, 'all event listeners were added');
+    assert.equal(removeEventListener.callCount, 4, 'all event listeners were removed');
+  });
+  test('error actions are triggered when errors are dispatched', async function(assert) {
+    const source = new RealEventSource();
+    const error = this.stub();
+    const close = this.stub();
+    const fakeService = Service.extend({
+      open: function(uri, obj) {
+        source.getCurrentEvent = function() {
+          return {};
+        };
+        return source;
+      },
+      close: close,
+    });
+    this.owner.register('service:data-source/fake-service', fakeService);
+    this.owner.inject('component:data-source', 'data', 'service:data-source/fake-service');
+    this.actions.change = data => {
+      source.dispatchEvent({ type: 'error', error: {} });
+    };
+    this.actions.error = error;
+    await render(
+      hbs`<DataSource @src="" @onchange={{action "change" value="data"}} @onerror={{action "error" value="error"}} />`
+    );
+    await waitUntil(() => {
+      return error.calledOnce;
+    });
+    assert.ok(error.calledOnce, 'error action was called');
+    assert.ok(close.calledOnce, 'close was called before the open');
+    await clearRender();
+    assert.ok(close.calledTwice, 'close was also called when the component is destroyed');
+  });
+});
diff --git a/ui-v2/tests/steps.js b/ui-v2/tests/steps.js
index 5d530fc66..b72cc6e82 100644
--- a/ui-v2/tests/steps.js
+++ b/ui-v2/tests/steps.js
@@ -50,20 +50,33 @@ const mb = function(path) {
   };
 };
 export default function(assert, library) {
-  const pauseUntil = function(cb) {
-    return new Promise(function(resolve, reject) {
+  const pauseUntil = function(run, message = 'assertion timed out') {
+    return new Promise(function(r, reject) {
       let count = 0;
-      const interval = setInterval(function() {
-        if (++count >= 50) {
-          clearInterval(interval);
-          assert.ok(false);
-          reject();
-        }
-        cb(function() {
-          clearInterval(interval);
-          resolve();
+      let resolved = false;
+      const retry = function() {
+        return Promise.resolve();
+      };
+      const resolve = function(str = message) {
+        resolved = true;
+        assert.ok(resolved, str);
+        r();
+        return Promise.resolve();
+      };
+      (function tick() {
+        run(resolve, reject, retry).then(function() {
+          if (!resolved) {
+            setTimeout(function() {
+              if (++count >= 50) {
+                assert.ok(false, message);
+                reject();
+                return;
+              }
+              tick();
+            }, 100);
+          }
         });
-      }, 100);
+      })();
     });
   };
   const lastNthRequest = getLastNthRequest(api.server.history);
diff --git a/ui-v2/tests/steps/assertions/dom.js b/ui-v2/tests/steps/assertions/dom.js
index 4716a2cfd..576efe798 100644
--- a/ui-v2/tests/steps/assertions/dom.js
+++ b/ui-v2/tests/steps/assertions/dom.js
@@ -1,16 +1,17 @@
 export default function(scenario, assert, pauseUntil, find, currentURL, clipboard) {
   scenario
     .then('pause until I see the text "$text" in "$selector"', function(text, selector) {
-      return pauseUntil(function(resolve) {
+      return pauseUntil(function(resolve, reject, retry) {
         const $el = find(selector);
         if ($el) {
           const hasText = $el.textContent.indexOf(text) !== -1;
           if (hasText) {
-            assert.ok(hasText, `Expected to see "${text}" in "${selector}"`);
-            resolve();
+            return resolve();
           }
+          return reject();
         }
-      });
+        return retry();
+      }, `Expected to see "${text}" in "${selector}"`);
     })
     .then(['I see the text "$text" in "$selector"'], function(text, selector) {
       const textContent = find(selector).textContent;
diff --git a/ui-v2/tests/steps/assertions/model.js b/ui-v2/tests/steps/assertions/model.js
index a42066f1c..bcacb8c0a 100644
--- a/ui-v2/tests/steps/assertions/model.js
+++ b/ui-v2/tests/steps/assertions/model.js
@@ -1,15 +1,15 @@
 export default function(scenario, assert, find, currentPage, pauseUntil, pluralize) {
   scenario
     .then('pause until I see $number $model model[s]?', function(num, model) {
-      return pauseUntil(function(resolve) {
+      return pauseUntil(function(resolve, reject, retry) {
         const len = currentPage()[pluralize(model)].filter(function(item) {
           return item.isVisible;
         }).length;
         if (len === num) {
-          assert.equal(len, num, `Expected ${num} ${model}s, saw ${len}`);
-          resolve();
+          return resolve();
         }
-      });
+        return retry();
+      }, `Expected ${num} ${model}s`);
     })
     .then(['I see $num $model model[s]?'], function(num, model) {
       const len = currentPage()[pluralize(model)].filter(function(item) {
diff --git a/ui-v2/tests/unit/services/data-source/protocols/http-test.js b/ui-v2/tests/unit/services/data-source/protocols/http-test.js
new file mode 100644
index 000000000..f861958d2
--- /dev/null
+++ b/ui-v2/tests/unit/services/data-source/protocols/http-test.js
@@ -0,0 +1,12 @@
+import { module, test } from 'qunit';
+import { setupTest } from 'ember-qunit';
+
+module('Unit | Service | data-source/protocols/http', function(hooks) {
+  setupTest(hooks);
+
+  // Replace this with your real tests.
+  test('it exists', function(assert) {
+    let service = this.owner.lookup('service:data-source/protocols/http');
+    assert.ok(service);
+  });
+});
diff --git a/ui-v2/tests/unit/services/data-source/protocols/local-storage-test.js b/ui-v2/tests/unit/services/data-source/protocols/local-storage-test.js
new file mode 100644
index 000000000..29920608f
--- /dev/null
+++ b/ui-v2/tests/unit/services/data-source/protocols/local-storage-test.js
@@ -0,0 +1,12 @@
+import { module, test } from 'qunit';
+import { setupTest } from 'ember-qunit';
+
+module('Unit | Service | data-source/protocols/local-storage', function(hooks) {
+  setupTest(hooks);
+
+  // Replace this with your real tests.
+  test('it exists', function(assert) {
+    let service = this.owner.lookup('service:data-source/protocols/local-storage');
+    assert.ok(service);
+  });
+});
diff --git a/ui-v2/tests/unit/utils/maybe-call-test.js b/ui-v2/tests/unit/utils/maybe-call-test.js
new file mode 100644
index 000000000..614d3f5da
--- /dev/null
+++ b/ui-v2/tests/unit/utils/maybe-call-test.js
@@ -0,0 +1,18 @@
+import maybeCall from 'consul-ui/utils/maybe-call';
+import { module, test } from 'qunit';
+import { Promise } from 'rsvp';
+
+module('Unit | Utility | maybe-call', function() {
+  test('it calls a function when the resolved value is true', function(assert) {
+    assert.expect(1);
+    return maybeCall(() => {
+      assert.ok(true);
+    }, Promise.resolve(true))();
+  });
+  test("it doesn't call a function when the resolved value is false", function(assert) {
+    assert.expect(0);
+    return maybeCall(() => {
+      assert.ok(true);
+    }, Promise.resolve(false))();
+  });
+});
diff --git a/ui-v2/yarn.lock b/ui-v2/yarn.lock
index 174c29d06..8efec6add 100644
--- a/ui-v2/yarn.lock
+++ b/ui-v2/yarn.lock
@@ -7311,7 +7311,7 @@ growly@^1.3.0:
   resolved "https://registry.yarnpkg.com/growly/-/growly-1.3.0.tgz#f10748cbe76af964b7c96c93c6bcc28af120c081"
   integrity sha1-8QdIy+dq+WS3yWyTxrzCivEgwIE=
 
-handlebars@^4.0.11, handlebars@^4.3.1, handlebars@^4.5.1:
+handlebars@^4.0.11, handlebars@^4.5.1:
   version "4.7.3"
   resolved "https://registry.yarnpkg.com/handlebars/-/handlebars-4.7.3.tgz#8ece2797826886cf8082d1726ff21d2a022550ee"
   integrity sha512-SRGwSYuNfx8DwHD/6InAPzD6RgeruWLT+B8e8a7gGs8FWgHzlExpTFMEq2IA6QpAfOClpKHy6+8IqTjeBCu6Kg==
@@ -7333,6 +7333,17 @@ handlebars@^4.0.4, handlebars@^4.1.2:
   optionalDependencies:
     uglify-js "^3.1.4"
 
+handlebars@^4.3.1:
+  version "4.7.2"
+  resolved "https://registry.yarnpkg.com/handlebars/-/handlebars-4.7.2.tgz#01127b3840156a0927058779482031afe0e730d7"
+  integrity sha512-4PwqDL2laXtTWZghzzCtunQUTLbo31pcCJrd/B/9JP8XbhVzpS5ZXuKqlOzsd1rtcaLo4KqAn8nl8mkknS4MHw==
+  dependencies:
+    neo-async "^2.6.0"
+    optimist "^0.6.1"
+    source-map "^0.6.1"
+  optionalDependencies:
+    uglify-js "^3.1.4"
+
 handlebars@~4.1.2:
   version "4.1.2"
   resolved "https://registry.yarnpkg.com/handlebars/-/handlebars-4.1.2.tgz#b6b37c1ced0306b221e094fc7aca3ec23b131b67"
@@ -9365,6 +9376,13 @@ mktemp@~0.4.0:
   resolved "https://registry.yarnpkg.com/mktemp/-/mktemp-0.4.0.tgz#6d0515611c8a8c84e484aa2000129b98e981ff0b"
   integrity sha1-bQUVYRyKjITkhKogABKbmOmB/ws=
 
+mnemonist@^0.30.0:
+  version "0.30.0"
+  resolved "https://registry.yarnpkg.com/mnemonist/-/mnemonist-0.30.0.tgz#c8c37f4425b8abcf7aa04a34199af254b398a90f"
+  integrity sha512-g9rbX4ug7am8oW3jnM7adaFaj5vv5PeOaMPNUwjKQQaTyY+qPQHTmUF59/ZOfQdtTknkjA7+7RhtmL2C0mtwPA==
+  dependencies:
+    obliterator "^1.5.0"
+
 morgan@^1.9.1:
   version "1.9.1"
   resolved "https://registry.yarnpkg.com/morgan/-/morgan-1.9.1.tgz#0a8d16734a1d9afbc824b99df87e738e58e2da59"
@@ -9809,6 +9827,11 @@ object.pick@^1.3.0:
   dependencies:
     isobject "^3.0.1"
 
+obliterator@^1.5.0:
+  version "1.5.0"
+  resolved "https://registry.yarnpkg.com/obliterator/-/obliterator-1.5.0.tgz#f3535e5be192473ef59efb2d30396738f7c645c6"
+  integrity sha512-dENe0UviDf8/auXn0bIBKwCcUr49khvSBWDLlszv/ZB2qz1VxWDmkNKFqO2nfmve7hQb/QIDY7+rc7K3LdJimQ==
+
 on-finished@~2.3.0:
   version "2.3.0"
   resolved "https://registry.yarnpkg.com/on-finished/-/on-finished-2.3.0.tgz#20f1336481b083cd75337992a16971aa2d906947"