feat(live): progress on panel <-> data source communication patterns

public/app/core/core.ts

@@ -29,6 +29,7 @@ import {colorPicker} from './components/colorpicker';
 import {navbarDirective} from './components/navbar/navbar';
 import {arrayJoin} from './directives/array_join';
 import {liveSrv} from './live/live_srv';
+import {Emitter} from './utils/emitter';
 import {layoutSelector} from './components/layout_selector/layout_selector';
 import 'app/core/controllers/all';
 import 'app/core/services/all';
@@ -46,5 +47,6 @@ export {
   colorPicker,
   liveSrv,
   layoutSelector,
-  infoPopover
+  infoPopover,
+  Emitter
 };

public/app/core/directives/plugin_component.ts

@@ -146,11 +146,11 @@ function pluginDirectiveLoader($compile, datasourceSrv, $rootScope, $q, $http, $
           };
         });
       }
-      // ConfigCtrl
+      // Datasource ConfigCtrl
       case 'datasource-config-ctrl': {
         var dsMeta = scope.ctrl.datasourceMeta;
         return System.import(dsMeta.module).then(function(dsModule): any {
-          if (!dsMeta.ConfigCtrl) {
+          if (!dsModule.ConfigCtrl) {
             return {notFound: true};
           }
 

public/app/core/utils/emitter.ts

@@ -0,0 +1,47 @@
+
+import {Subject} from 'vendor/npm/rxjs/Subject';
+
+var hasOwnProp = {}.hasOwnProperty;
+
+function createName(name) {
+    return '$' + name;
+}
+
+export class Emitter {
+  subjects: any;
+
+  constructor() {
+    this.subjects = {};
+  }
+
+  emit(name, data) {
+    var fnName = createName(name);
+    this.subjects[fnName] || (this.subjects[fnName] = new Subject());
+    this.subjects[fnName].next(data);
+  }
+
+  on(name, handler) {
+    var fnName = createName(name);
+    this.subjects[fnName] || (this.subjects[fnName] = new Subject());
+    this.subjects[fnName].subscribe(handler);
+  };
+
+  off(name, handler) {
+    var fnName = createName(name);
+    if (this.subjects[fnName]) {
+      this.subjects[fnName].dispose();
+      delete this.subjects[fnName];
+    }
+  }
+
+  dispose() {
+    var subjects = this.subjects;
+    for (var prop in subjects) {
+      if (hasOwnProp.call(subjects, prop)) {
+        subjects[prop].dispose();
+      }
+    }
+
+    this.subjects = {};
+  }
+}

public/app/features/panel/metrics_panel_ctrl.ts

@@ -9,6 +9,8 @@ import {PanelCtrl} from './panel_ctrl';
 import * as rangeUtil from 'app/core/utils/rangeutil';
 import * as dateMath from 'app/core/utils/datemath';
 
+import {Subject} from 'vendor/npm/rxjs/Subject';
+
 class MetricsPanelCtrl extends PanelCtrl {
   error: boolean;
   loading: boolean;
@@ -26,7 +28,8 @@ class MetricsPanelCtrl extends PanelCtrl {
   timeInfo: any;
   skipDataOnInit: boolean;
   datasources: any[];
-  dataSubject: any;
+  dataStream: any;
+  dataSubscription: any;
 
   constructor($scope, $injector) {
     super($scope, $injector);
@@ -50,11 +53,6 @@ class MetricsPanelCtrl extends PanelCtrl {
     this.datasources = this.datasourceSrv.getMetricSources();
   }
 
-  refreshData(data) {
-    // null op
-    return this.$q.when(data);
-  }
-
   loadSnapshot(data) {
     // null op
     return data;
@@ -73,21 +71,27 @@ class MetricsPanelCtrl extends PanelCtrl {
       return;
     }
 
+    // // ignore if we have data stream
+    if (this.dataStream) {
+      return;
+    }
+
     // clear loading/error state
     delete this.error;
     this.loading = true;
 
     // load datasource service
-    this.datasourceSrv.get(this.panel.datasource).then(datasource => {
+    this.datasourceSrv.get(this.panel.datasource)
-      this.datasource = datasource;
+    .then(this.issueQueries.bind(this))
-      return this.refreshData(this.datasource);
+    .then(() => {
-    }).then(() => {
       this.loading = false;
     }).catch(err => {
       console.log('Panel data error:', err);
       this.loading = false;
       this.error = err.message || "Timeseries data request error";
       this.inspector = {error: err};
+
+      this.events.emit('data-error', err);
     });
   }
 
@@ -167,10 +171,6 @@ class MetricsPanelCtrl extends PanelCtrl {
       return this.$q.when([]);
     }
 
-    if (this.dataSubject) {
-      return this.$q.when([]);
-    }
-
     var metricsQuery = {
       panelId: this.panel.id,
       range: this.range,
@@ -190,15 +190,15 @@ class MetricsPanelCtrl extends PanelCtrl {
 
         // check for if data source returns subject
         if (results && results.subscribe) {
-          this.handleDataSubject(results);
+          this.handleDataStream(results);
-          return {data: []};
+          return;
         }
 
         if (this.dashboard.snapshot) {
           this.panel.snapshotData = results;
         }
 
-        return this.dataHandler(results);
+        return this.events.emit('data-received', results);
       });
     } catch (err) {
       return this.$q.reject(err);
@@ -209,22 +209,24 @@ class MetricsPanelCtrl extends PanelCtrl {
     return data;
   }
 
-  handleDataSubject(subject) {
+  handleDataStream(stream) {
     // if we already have a connection
-    if (this.dataSubject) {
+    if (this.dataStream) {
+      console.log('two stream observables!');
       return;
     }
 
-    this.dataSubject = subject;
+    this.dataStream = stream;
-    this.dataSubject.subscribe({
+    this.dataSubscription = stream.subscribe({
       next: (data) => {
         console.log('dataSubject next!');
         if (data.range) {
           this.range = data.range;
         }
-        this.dataHandler(data);
+        this.events.emit('data-received', data);
       },
       error: (error) => {
+        this.events.emit('data-error', error);
         console.log('panel: observer got error');
       },
       complete: () => {

public/app/features/panel/panel_ctrl.ts

@@ -9,6 +9,8 @@ const TITLE_HEIGHT = 25;
 const EMPTY_TITLE_HEIGHT = 9;
 const PANEL_PADDING = 5;
 
+import {Emitter} from 'app/core/core';
+
 export class PanelCtrl {
   panel: any;
   row: any;
@@ -28,12 +30,14 @@ export class PanelCtrl {
   editMode: any;
   height: any;
   containerHeight: any;
+  events: Emitter;
 
   constructor($scope, $injector) {
     this.$injector = $injector;
     this.$scope = $scope;
     this.$timeout = $injector.get('$timeout');
     this.editorTabIndex = 0;
+    this.events = new Emitter();
 
     var plugin = config.panels[this.panel.type];
     if (plugin) {
@@ -56,7 +60,7 @@ export class PanelCtrl {
   }
 
   refresh() {
-    return;
+    this.render();
   }
 
   publishAppEvent(evtName, evt) {
@@ -138,7 +142,7 @@ export class PanelCtrl {
     this.height = this.containerHeight - (PANEL_PADDING + (this.panel.title ? TITLE_HEIGHT : EMPTY_TITLE_HEIGHT));
   }
 
-  broadcastRender(arg1?, arg2?) {
+  render(arg1?, arg2?) {
     this.$scope.$broadcast('render', arg1, arg2);
   }
 
@@ -157,7 +161,7 @@ export class PanelCtrl {
   updateColumnSpan(span) {
     this.panel.span = Math.min(Math.max(Math.floor(this.panel.span + span), 1), 12);
     this.$timeout(() => {
-      this.broadcastRender();
+      this.render();
     });
   }
 

public/app/plugins/datasource/grafana-live/datasource.ts

@@ -24,7 +24,6 @@ export class GrafanaStreamDS {
 
   /** @ngInject */
   constructor() {
-
   }
 
   query(options): any {

public/app/plugins/panel/graph/module.ts

@@ -106,6 +106,9 @@ class GraphCtrl extends MetricsPanelCtrl {
     _.defaults(this.panel.legend, panelDefaults.legend);
 
     this.colors = $scope.$root.colors;
+
+    this.events.on('data-received', this.onDataReceived.bind(this));
+    this.events.on('data-error', this.onDataError.bind(this));
   }
 
   initEditMode() {
@@ -138,14 +141,9 @@ class GraphCtrl extends MetricsPanelCtrl {
     this.render();
   }
 
-  refreshData(datasource) {
+  issueQueries(datasource) {
     this.annotationsPromise = this.annotationsSrv.getAnnotations(this.dashboard);
-
+    return super.issueQueries(datasource);
-    return this.issueQueries(datasource).catch(err => {
-      this.seriesList = [];
-      this.render([]);
-      throw err;
-    });
   }
 
   zoomOut(evt) {
@@ -157,7 +155,12 @@ class GraphCtrl extends MetricsPanelCtrl {
     this.dataHandler(snapshotData);
   }
 
-  dataHandler(results) {
+  onDataError(err) {
+    this.seriesList = [];
+    this.render([]);
+  }
+
+  onDataReceived(results) {
     // png renderer returns just a url
     if (_.isString(results)) {
       this.render(results);
@@ -178,7 +181,7 @@ class GraphCtrl extends MetricsPanelCtrl {
       this.loading = false;
       this.render(this.seriesList);
     });
-  };
+  }
 
   seriesHandler(seriesData, index) {
     var datapoints = seriesData.datapoints;
@@ -208,10 +211,6 @@ class GraphCtrl extends MetricsPanelCtrl {
     return series;
   }
 
-  render(data?: any) {
-    this.broadcastRender(data);
-  }
-
   changeSeriesColor(series, color) {
     series.color = color;
     this.panel.aliasColors[series.alias] = series.color;

public/app/plugins/panel/table/module.ts

@@ -80,9 +80,7 @@ class TablePanelCtrl extends MetricsPanelCtrl {
       });
     }
 
-    return this.issueQueries(datasource)
+    return this.issueQueries(datasource).catch(err => {
-    .then(this.dataHandler.bind(this))
-    .catch(err => {
       this.render();
       throw err;
     });

public/app/plugins/panel/text/module.ts

@@ -29,11 +29,9 @@ export class TextPanelCtrl extends PanelCtrl {
     this.editorTabIndex = 1;
   }
 
-  refresh() {
-    this.render();
-  }
-
   render() {
+    super.render();
+
     if (this.panel.mode === 'markdown') {
       this.renderMarkdown(this.panel.content);
     } else if (this.panel.mode === 'html') {

public/app/plugins/panel/unknown/module.ts

@@ -9,6 +9,7 @@ export class UnknownPanelCtrl extends PanelCtrl {
   constructor($scope, $injector) {
     super($scope, $injector);
   }
+
 }