Timefilter Auto Refresh - Closes "Support for auto-refresh #1845"

src/kibana/components/courier/_request_queue.js

@@ -0,0 +1,35 @@
+define(function (require) {
+  return function PendingRequestList() {
+    var _ = require('lodash');
+
+    /**
+     * Queue of pending requests, requests are removed as
+     * they are processed by fetch.[sourceType]().
+     * @type {Array}
+     */
+    var queue = [];
+
+    queue.getInactive = function (/* strategies */) {
+      return queue.get.apply(queue, arguments)
+      .filter(function (req) {
+        return !req.started;
+      });
+    };
+
+    queue.get = function (/* strategies.. */) {
+      var strategies = _.toArray(arguments);
+      return queue.filter(function (req) {
+        var strategyMatch = !strategies.length;
+        if (!strategyMatch) {
+          strategyMatch = strategies.some(function (strategy) {
+            return req.strategy === strategy;
+          });
+        }
+
+        return strategyMatch && req.canStart();
+      });
+    };
+
+    return queue;
+  };
+});

src/kibana/components/courier/courier.js

@@ -1,20 +1,23 @@
 define(function (require) {
   var errors = require('errors');
-
+  var _ = require('lodash');
+
   require('services/es');
   require('services/promises');
   require('components/index_patterns/index_patterns');
 
   require('modules').get('kibana/courier')
-  .service('courier', function ($rootScope, Private, Promise, indexPatterns) {
+  .service('courier', function ($rootScope, Private, Promise, indexPatterns, Notifier) {
     function Courier() {
       var self = this;
 
       var DocSource = Private(require('components/courier/data_source/doc_source'));
       var SearchSource = Private(require('components/courier/data_source/search_source'));
 
-      var pendingRequests = Private(require('components/courier/_pending_requests'));
+      var requestQueue = Private(require('components/courier/_request_queue'));
+      var errorHandlers = Private(require('components/courier/_error_handlers'));
 
+      var fetch = Private(require('components/courier/fetch/fetch'));
       var docLooper = self.docLooper = Private(require('components/courier/looper/doc'));
       var searchLooper = self.searchLooper = Private(require('components/courier/looper/search'));
 
@@ -29,7 +32,6 @@ define(function (require) {
       self.SearchSource = SearchSource;
 
       var HastyRefresh = errors.HastyRefresh;
-      var Abort = errors.Abort;
 
       /**
        * update the time between automatic search requests
@@ -47,6 +49,7 @@ define(function (require) {
        */
       self.start = function () {
         searchLooper.start();
+        docLooper.start();
         return this;
       };
 
@@ -56,7 +59,9 @@ define(function (require) {
        * individual errors are routed to their respective requests.
        */
       self.fetch = function () {
-        return searchLooper.run();
+        fetch.searches().then(function () {
+          searchLooper.restart();
+        });
       };
 
 
@@ -105,15 +110,26 @@ define(function (require) {
         searchLooper.stop();
         docLooper.stop();
 
-        [].concat(pendingRequests.splice(0), this._errorHandlers.splice(0))
-        .forEach(function (req) {
-          req.defer.reject(new Abort());
-        });
+        _.invoke(requestQueue, 'abort');
 
-        if (pendingRequests.length) {
+        if (requestQueue.length) {
           throw new Error('Aborting all pending requests failed.');
         }
       };
+
+      // Listen for refreshInterval changes
+      $rootScope.$watch('timefilter.refreshInterval', function () {
+        var refreshValue = _.deepGet($rootScope, 'timefilter.refreshInterval.value');
+        if (_.isNumber(refreshValue)) {
+          self.fetchInterval(refreshValue);
+        } else {
+          self.fetchInterval(0);
+        }
+      });
+
+      var onFatalDefer = Promise.defer();
+      onFatalDefer.promise.then(self.close);
+      Notifier.fatalCallbacks.push(onFatalDefer.resolve);
     }
 
     return new Courier();

src/kibana/components/courier/data_source/_abstract.js

@@ -1,10 +1,8 @@
 define(function (require) {
-  var inherits = require('lodash').inherits;
   var _ = require('lodash');
-  var nextTick = require('utils/next_tick');
 
-  return function SourceAbstractFactory(Private, Promise, PromiseEmitter, timefilter) {
-    var pendingRequests = Private(require('components/courier/_pending_requests'));
+  return function SourceAbstractFactory(Private, Promise, PromiseEmitter) {
+    var requestQueue = Private(require('components/courier/_request_queue'));
     var errorHandlers = Private(require('components/courier/_error_handlers'));
     var courierFetch = Private(require('components/courier/fetch/fetch'));
 
@@ -123,8 +121,7 @@ define(function (require) {
       var self = this;
 
       return new PromiseEmitter(function (resolve, reject, defer) {
-        var req = self._createRequest(defer);
-        pendingRequests.push(req);
+        self._createRequest(defer);
       }, handler);
     };
 
@@ -154,64 +151,61 @@ define(function (require) {
 
     /**
      * Fetch just this source ASAP
-     * @param {Function} cb - callback
+     *
+     * ONLY USE IF YOU WILL BE USING THE RESULTS
+     * provided by the returned promise, otherwise
+     * call #fetchQueued()
+     *
+     * @async
      */
     SourceAbstract.prototype.fetch = function () {
       var self = this;
+      var req = _.first(self._myQueued());
 
-      var req = self._createRequest();
-      pendingRequests.push(req);
-
-      // fetch just the requests for this source
-      courierFetch.these(self._getType(), pendingRequests.splice(0).filter(function (req) {
-        if (req.source !== self) {
-          pendingRequests.push(req);
-          return false;
-        }
+      if (!req) {
+        req = self._createRequest();
+      }
 
-        return true;
-      }));
+      courierFetch.these([req]);
 
       return req.defer.promise;
     };
 
+    /**
+     * Fetch all pending requests for this source ASAP
+     * @async
+     */
+    SourceAbstract.prototype.fetchQueued = function () {
+      return courierFetch.these(this._myQueued());
+    };
+
     /**
      * Cancel all pending requests for this dataSource
      * @return {undefined}
      */
-    SourceAbstract.prototype.cancelPending = function () {
-      var pending = _.where(pendingRequests, { source: this});
-      _.pull.apply(_, [pendingRequests].concat(pending));
+    SourceAbstract.prototype.cancelQueued = function () {
+      _.invoke(this._myQueued(), 'abort');
     };
 
     /**
      * Completely destroy the SearchSource.
      * @return {undefined}
      */
     SourceAbstract.prototype.destroy = function () {
-      this.cancelPending();
+      this.cancelQueued();
     };
 
     /*****
      * PRIVATE API
      *****/
 
-    SourceAbstract.prototype._createRequest = function (defer) {
-      var self = this;
-
-      var req = {
-        source: self,
-        defer: defer || Promise.defer()
-      };
-
-      if (self.history) {
-        // latest history at the top
-        self.history.unshift(req);
-        // trim all entries beyond 19/20
-        self.history.splice(20);
-      }
+    SourceAbstract.prototype._myQueued = function () {
+      var reqs = requestQueue.get(this._fetchStrategy);
+      return _.where(reqs, { source: this });
+    };
 
-      return req;
+    SourceAbstract.prototype._createRequest = function () {
+      throw new Error('_createRequest must be implemented by subclass');
     };
 
     /**
@@ -233,7 +227,7 @@ define(function (require) {
       var current = this;
 
       // call the ittr and return it's promise
-      return (function ittr(resolve, reject) {
+      return (function ittr() {
         // itterate the _state object (not array) and
         // pass each key:value pair to source._mergeProp. if _mergeProp
         // returns a promise, then wait for it to complete and call _mergeProp again
@@ -259,6 +253,8 @@ define(function (require) {
       }())
       .then(function () {
         if (type === 'search') {
+          flatState.body = flatState.body || {};
+
           // defaults for the query
           if (!flatState.body.query) {
             flatState.body.query = {

src/kibana/components/courier/data_source/_doc_send_to_es.js

@@ -3,7 +3,8 @@ define(function (require) {
   var errors = require('errors');
 
   return function (Promise, Private, es) {
-    var pendingRequests = Private(require('components/courier/_pending_requests'));
+    var requestQueue = Private(require('components/courier/_request_queue'));
+    var courierFetch = Private(require('components/courier/fetch/fetch'));
 
     /**
      * Backend for doUpdate and doIndex
@@ -44,19 +45,22 @@ define(function (require) {
 
           // clear the queue and filter out the removed items, pushing the
           // unmatched ones back in.
-          pendingRequests.splice(0).filter(function (req) {
+          var respondTo = requestQueue.splice(0).filter(function (req) {
             var isDoc = req.source._getType() === 'doc';
             var keyMatches = isDoc && req.source._versionKey() === key;
 
-            if (keyMatches) {
-              // resolve the request with a copy of the response
-              req.defer.resolve(_.cloneDeep(fetchResp));
-              return;
+            // put some request back into the queue
+            if (!keyMatches) {
+              requestQueue.push(req);
+              return false;
             }
 
-            // otherwise, put the request back into the queue
-            pendingRequests.push(req);
+            return true;
           });
+
+          return courierFetch.fakeFetchThese(respondTo, respondTo.map(function () {
+            return _.cloneDeep(fetchResp);
+          }));
         });
 
         return resp._id;

src/kibana/components/courier/data_source/_root_search_source.js

@@ -1,6 +1,5 @@
 define(function (require) {
   return function RootSearchSource(Private, $rootScope, config, Promise, indexPatterns, timefilter, Notifier) {
-    var _ = require('lodash');
     var SearchSource = Private(require('components/courier/data_source/search_source'));
 
     var notify = new Notifier({ location: 'Root Search Source' });

src/kibana/components/courier/data_source/doc_source.js

@@ -1,15 +1,10 @@
 define(function (require) {
   var _ = require('lodash');
-  var errors = require('errors');
-
-  var inherits = require('lodash').inherits;
 
   return function DocSourceFactory(Private, Promise, es, sessionStorage) {
     var sendToEs = Private(require('components/courier/data_source/_doc_send_to_es'));
     var SourceAbstract = Private(require('components/courier/data_source/_abstract'));
-
-    var VersionConflict = errors.VersionConflict;
-    var RequestFailure = errors.RequestFailure;
+    var DocRequest = Private(require('components/courier/fetch/request/doc'));
 
     _(DocSource).inherits(SourceAbstract);
     function DocSource(initialState) {
@@ -23,6 +18,10 @@ define(function (require) {
      * PUBLIC API
      *****/
 
+    DocSource.prototype._createRequest = function (defer) {
+      return new DocRequest(this, defer);
+    };
+
     /**
      * List of methods that is turned into a chainable API in the constructor
      * @type {Array}

src/kibana/components/courier/data_source/search_source.js

@@ -2,11 +2,9 @@ define(function (require) {
 
   return function SearchSourceFactory(Promise, Private) {
     var _ = require('lodash');
-    var errors = require('errors');
     var SourceAbstract = Private(require('components/courier/data_source/_abstract'));
-
-    var FetchFailure = errors.FetchFailure;
-    var RequestFailure = errors.RequestFailure;
+    var SearchRequest = Private(require('components/courier/fetch/request/search'));
+    var SegmentedRequest = Private(require('components/courier/fetch/request/segmented'));
 
     _(SearchSource).inherits(SourceAbstract);
     function SearchSource(initialState) {
@@ -125,6 +123,18 @@ define(function (require) {
       return normal;
     };
 
+    SearchSource.prototype.onBeginSegmentedFetch = function (initFunction) {
+      var self = this;
+      return Promise.try(function addRequest() {
+        var req = new SegmentedRequest(self, Promise.defer(), initFunction);
+
+        // return promises created by the completion handler so that
+        // errors will bubble properly
+        return req.defer.promise.then(addRequest);
+      });
+    };
+
+
     /******
      * PRIVATE APIS
      ******/
@@ -137,6 +147,19 @@ define(function (require) {
       return 'search';
     };
 
+    /**
+     * Create a common search request object, which should
+     * be put into the pending request queye, for this search
+     * source
+     *
+     * @param {Deferred} defer - the deferred object that should be resolved
+     *                         when the request is complete
+     * @return {SearchRequest}
+     */
+    SearchSource.prototype._createRequest = function (defer) {
+      return new SearchRequest(this, defer);
+    };
+
     /**
      * Used to merge properties into the state within ._flatten().
      * The state is passed in and modified by the function