reverted code for elasticsearch api revert resolves #579

lib/processors/elasticsearch_index_selector.js

@@ -46,19 +46,15 @@ function newProcessor(context, opConfig, jobConfig) {
      * id_field - string. Name of the field that contains the id to use for the inserted record.
      * preserve_id - boolean. if incoming data is from ES the existing ID of the record will be preserved.
      */
-    function isFullResponse(array) {
-        const record = array[0];
-        if (record && record._source && record._id && record._index) {
-            return true;
-        }
-        return false;
-    }
 
     return function(data) {
         var fromElastic = false;
+        var dataArray = data;
+        var fullResponseData = _.get(dataArray, 'hits.hits');
 
-        if (isFullResponse(data)) {
+        if (fullResponseData) {
             fromElastic = true;
+            dataArray = fullResponseData;
         }
 
         if (!opConfig.type && !fromElastic) {
@@ -69,27 +65,27 @@ function newProcessor(context, opConfig, jobConfig) {
         function generateRequest(start) {
             var record;
             if (fromElastic) {
-                record = data[start]._source;
+                record = dataArray[start]._source;
             } else {
-                record = data[start];
+                record = dataArray[start];
             }
 
             var indexSpec = {};
 
             var meta = {
                 "_index": indexName(record),
-                "_type": opConfig.type ? opConfig.type : data[start]._type
+                "_type": opConfig.type ? opConfig.type : data.hits.hits[start]._type
             };
 
             if (opConfig.preserve_id) {
-                meta._id = data[start]._id;
+                meta._id = data.hits.hits[start]._id;
             }
             else if (opConfig.id_field) {
                 if (fromElastic) {
-                    meta._id = data[start]._source[opConfig.id_field];
+                    meta._id = dataArray[start]._source[opConfig.id_field];
                 }
                 else {
-                    meta._id = data[start][opConfig.id_field];
+                    meta._id = dataArray[start][opConfig.id_field];
                 }
             }
 
@@ -160,7 +156,7 @@ function newProcessor(context, opConfig, jobConfig) {
             }
         }
 
-        for (var i = 0; i < data.length; i++) {
+        for (var i = 0; i < dataArray.length; i++) {
             generateRequest(i)
         }
 

lib/readers/elasticsearch_date_range/slicer.js

@@ -118,7 +118,7 @@ function newSlicer(context, opConfig, job, retryData, logger, client) {
         //using this query to catch potential errors even if a date is given already
         return elasticsearch.search(query)
             .then(function(results) {
-                var data = _.get(results, '[0]._source', results[0]);
+                var data = _.get(results, 'hits.hits[0]._source', results[0]);
                 if (data === undefined) {
                     return null
                 }

spec/processors/elasticsearch_index_selector-spec.js

@@ -87,7 +87,7 @@ describe('elasticsearch index selector', function() {
         var context = {};
         var opConfig = {index: 'someIndex', type: 'events', preserve_id: true, delete: false};
         var jobConfig = {logger: 'im a fake logger'};
-        var data = [{type: 'someType', _index: 'some_index', _id: 'specialID', _source: {some: 'data'}}];
+        var data = {hits: {hits: [{type: 'someType', _index: 'some_index', _id: 'specialID', _source: {some: 'data'}}]}};
 
         var fn = indexer.newProcessor(context, opConfig, jobConfig);
         var results = fn(data);