Merge pull request #1 from spenceralger/segment_in_courier

Segmented fetch in courier

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

@@ -154,34 +154,38 @@ define(function (require) {
 
     /**
      * Fetch just this source ASAP
-     * @param {Function} cb - callback
+     *
+     * ONLY USE IF YOU NEED THE RESULTS OTHERWISE USE .fetchPending()
+     * TO TRIGGER FETCHING ALL PENDING REQUESTS
+     *
+     * @async
      */
     SourceAbstract.prototype.fetch = function () {
       var self = this;
+      var req = _.first(self._myPending());
+      if (!req) {
+        req = self._createRequest();
+        pendingRequests.push(req);
+      }
 
-      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;
-        }
-
-        return true;
-      }));
-
+      self.fetchPending();
       return req.defer.promise;
     };
 
+    /**
+     * Fetch all pending requests for this source ASAP
+     * @async
+     */
+    SourceAbstract.prototype.fetchPending = function () {
+      return courierFetch.these(this._pullMyPending());
+    };
+
     /**
      * Cancel all pending requests for this dataSource
      * @return {undefined}
      */
     SourceAbstract.prototype.cancelPending = function () {
-      var pending = _.where(pendingRequests, { source: this});
-      _.pull.apply(_, [pendingRequests].concat(pending));
+      this._pullMyPending();
     };
 
     /**
@@ -196,12 +200,28 @@ define(function (require) {
      * PRIVATE API
      *****/
 
+    SourceAbstract.prototype._myPending = function () {
+      return _.where(pendingRequests, { source: this });
+    };
+
+    SourceAbstract.prototype._pullMyPending = function () {
+      var self = this;
+      return pendingRequests.splice(0).filter(function (req) {
+        if (req.source !== self) {
+          pendingRequests.push(req);
+          return false;
+        }
+        return true;
+      });
+    };
+
     SourceAbstract.prototype._createRequest = function (defer) {
       var self = this;
 
       var req = {
         source: self,
-        defer: defer || Promise.defer()
+        defer: defer || Promise.defer(),
+        strategy: self._fetchStrategy
       };
 
       if (self.history) {

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

@@ -4,6 +4,8 @@ define(function (require) {
     var _ = require('lodash');
     var errors = require('errors');
     var SourceAbstract = Private(require('components/courier/data_source/_abstract'));
+    var pendingRequests = Private(require('components/courier/_pending_requests'));
+    var segmentedStrategy = Private(require('components/courier/fetch/strategy/segmented'));
 
     var FetchFailure = errors.FetchFailure;
     var RequestFailure = errors.RequestFailure;
@@ -125,6 +127,25 @@ define(function (require) {
       return normal;
     };
 
+    SearchSource.prototype.onBeginSegmentedFetch = function (init) {
+      var self = this;
+      return Promise.try(function addRequest() {
+        var defer = Promise.defer();
+
+        var req = self._createRequest(defer);
+        // add a couple items to this request to identify it as a segmented fetch request
+        req.segmented = true;
+        req.strategy = segmentedStrategy;
+        req.init = init;
+
+        pendingRequests.push(req);
+
+        // return promises created by the completion handler so that
+        // errors will bubble properly
+        return defer.promise.then(addRequest);
+      });
+    };
+
     /******
      * PRIVATE APIS
      ******/

src/kibana/components/courier/fetch/_fetch_these.js

@@ -0,0 +1,151 @@
+define(function (require) {
+  return function FetchTheseProvider(Private, Promise, es, Notifier, sessionId, configFile) {
+    var _ = require('lodash');
+    var moment = require('moment');
+    var errors = require('errors');
+    var pendingRequests = Private(require('components/courier/_pending_requests'));
+
+    var notify = new Notifier({
+      location: 'Courier Fetch'
+    });
+
+    function eachStrategy(requests, block) {
+      block = Promise.method(block);
+      var sets = [];
+
+      requests.forEach(function (req) {
+        var strategy = req.strategy;
+        var set = _.find(sets, { 0: strategy });
+        if (set) set[1].push(req);
+        else sets.push([strategy, [req]]);
+      });
+
+      return Promise.all(sets.map(function (set) {
+        return (function fetch(requests, strategy) {
+          return block(requests, strategy)
+          .then(function checkForIncompleteRequests(result) {
+            if (_.isFunction(strategy.getIncompleteRequests)) {
+              var incomplete = strategy.getIncompleteRequests(pendingRequests);
+              if (incomplete.length) {
+                return fetch(incomplete, strategy);
+              }
+            }
+            return result;
+          });
+        }(set[1], set[0]));
+      }))
+      .catch(notify.fatal);
+    }
+
+    function initRequest(req) {
+      if (req.source.activeFetchCount) {
+        req.source.activeFetchCount += 1;
+      } else {
+        req.source.activeFetchCount = 1;
+      }
+
+      req.moment = moment();
+    }
+
+    function mergeDuplicateRequests(requests) {
+      // dedupe requests
+      var index = {};
+      return requests.splice(0).filter(function (req) {
+        var iid = req.source._instanceid;
+        if (!index[iid]) {
+          // this request is unique so far
+          index[iid] = req;
+          // keep the request
+          return true;
+        }
+
+        // the source was requested at least twice
+        var uniq = index[iid];
+        if (uniq._merged) {
+          // already setup the merged list
+          uniq._merged.push(req);
+        } else {
+          // put all requests into this array and itterate them on response
+          uniq._merged = [uniq, req];
+        }
+      });
+    }
+
+    function reqComplete(req, resp, errorHandler) {
+      if (resp.timed_out) {
+        notify.warning(new errors.SearchTimeout());
+      }
+
+      req.complete = true;
+      req.resp = resp;
+      req.ms = req.moment.diff() * -1;
+      req.source.activeFetchCount -= 1;
+
+      if (resp.error) {
+        return errorHandler.handle(req, new errors.FetchFailure(resp));
+      }
+
+      req.strategy.resolveRequest(req, resp);
+    }
+
+    function fetchThese(requests, errorHandler) {
+      return eachStrategy(requests, function (requests, strategy) {
+        requests.forEach(initRequest);
+
+        var uniq = mergeDuplicateRequests(requests);
+        var states;
+
+        return Promise.map(uniq, function (req) {
+          return strategy.getSourceStateFromRequest(req);
+        })
+        .then(function (_states_) {
+          states = _states_;
+
+          // all requests must have been disabled
+          if (!states.length) return Promise.resolve(false);
+
+          return es[strategy.clientMethod]({
+            timeout: configFile.shard_timeout,
+            preference: sessionId,
+            body: strategy.convertStatesToBody(states)
+          });
+        })
+        .then(strategy.getResponses)
+        .then(function (responses) {
+
+          responses.forEach(function (resp) {
+            var req = uniq.shift();
+            var state = states.shift();
+            if (!req._merged) {
+              req.state = state;
+              reqComplete(req, resp, errorHandler);
+            } else {
+              req._merged.forEach(function (mergedReq) {
+                mergedReq.state = state;
+                var respClone = _.cloneDeep(resp);
+                reqComplete(mergedReq, respClone, errorHandler);
+              });
+            }
+          });
+
+          return responses;
+        })
+        .catch(function (err) {
+
+          function sendFailure(req) {
+            req.source.activeFetchCount -= 1;
+            errorHandler.handle(req, err);
+          }
+
+          uniq.forEach(function (req) {
+            if (!req._merged) sendFailure(req);
+            else req._merged.forEach(sendFailure);
+          });
+          throw err;
+        });
+      });
+    }
+
+    return fetchThese;
+  };
+});