Use buffered queue simplify the queue operations (#81)

package.json

@@ -31,6 +31,7 @@
   "homepage": "https://github.com/salesforce/refocus-collector#readme",
   "dependencies": {
     "bluebird": "^3.5.0",
+    "buffered-queue": "^0.1.4",
     "coveralls": "^2.13.1",
     "debug": "^2.6.8",
     "errors": "^0.3.0",

src/commands/start.js

@@ -16,7 +16,6 @@ const debug = require('debug')('refocus-collector:commands');
 const logger = require('winston');
 const configModule = require('../config/config');
 const repeater = require('../repeater/repeater');
-const flush = require('../sampleQueue/sampleQueueOps').flush;
 const sendHeartbeat = require('../heartbeat/heartbeat').sendHeartbeat;
 const registryFile = require('../constants').registryLocation;
 const registryFileUtils = require('../utils/registryFileUtils');
@@ -45,13 +44,6 @@ function execute(name) {
     onProgress: debug,
   });
 
-  // flush function does not return anything, hence no event functions
-  repeater.create({
-    name: 'SampleQueueFlush',
-    interval: config.collectorConfig.sampleUpsertQueueTime,
-    func: () => flush(config.collectorConfig.maxSamplesPerBulkRequest,
-      regObj),
-  });
   logger.info({ activity: 'cmdStart' });
   debug('Exiting start.execute');
 } // execute

src/constants.js

@@ -32,6 +32,9 @@ module.exports = {
     SERVICE_UNAVAILABLE: 503,
   },
 
+  // Bulk upsert Sample Queue Name
+  bulkUpsertSampleQueue: 'bulkUpsertSampleQueue',
+
   // exported for the purpose of testing
   mockRegistryLocation: './test/config/testRegistry.json',
 

src/heartbeat/heartbeat.js

@@ -73,8 +73,8 @@ function sendHeartbeat(refocusInstanceObj) {
     };
 
     return buildMockResponse(generatorsDir)
-    .then(res => handleHeartbeatResponse(null, res))
-    .catch(err => handleHeartbeatResponse(err, null));
+    .then(res => handleHeartbeatResponse(null, res, refocusInstanceObj))
+    .catch(err => handleHeartbeatResponse(err, null, refocusInstanceObj));
 
     //TODO: send the real request and handle the response once the api can handle it
     //debug(`sendHeartbeat sending request. url: ${url} body: %o`, body);

src/heartbeat/listener.js

@@ -13,6 +13,9 @@ const debug = require('debug')('refocus-collector:heartbeat');
 const logger = require('winston');
 const utils = require('./utils');
 const configModule = require('../config/config');
+const queueUtils = require('../utils/queueUtils');
+const httpUtils = require('../utils/httpUtils');
+const bulkUpsertSampleQueue = require('../constants').bulkUpsertSampleQueue;
 
 /**
  * Handles the heartbeat response:
@@ -24,14 +27,36 @@ const configModule = require('../config/config');
  * @returns {Object} - config object. An error object is returned if this
  *  function is called with the error as the first argument.
  */
-function handleHeartbeatResponse(err, res) {
+function handleHeartbeatResponse(err, res, refocusInstanceObj=null) {
   debug('entered handleHeartbeatResponse');
   if (err) {
     logger.error('The handleHeartbeatResponse function was called with an ' +
       'error:', err);
     return err;
   }
 
+  // queue generation
+  // get queue
+  const _bulkUpsertSampleQueue = queueUtils.getQueue(bulkUpsertSampleQueue);
+  if (_bulkUpsertSampleQueue) {
+    if (res.collectorConfig) {
+      _bulkUpsertSampleQueue._size = res.collectorConfig.maxSamplesPerBulkRequest;
+      _bulkUpsertSampleQueue._flushTimeout =
+        res.collectorConfig.sampleUpsertQueueTime;
+    }
+  } else {
+    const config = configModule.getConfig();
+    const queueParams = {
+      name: bulkUpsertSampleQueue,
+      size: config.collectorConfig.maxSamplesPerBulkRequest,
+      flushTimeout: config.collectorConfig.sampleUpsertQueueTime,
+      verbose: false,
+      flushFunction: httpUtils.doBulkUpsert,
+      refocusInstanceObj: refocusInstanceObj,
+    };
+    queueUtils.createQueue(queueParams);
+  }
+
   if (res) {
     utils.updateCollectorConfig(res);
     utils.addGenerator(res);

src/remoteCollection/handleCollectResponse.js

@@ -14,8 +14,10 @@ const evalUtils = require('../utils/evalUtils');
 const errors = require('../errors');
 const errorSamples = require('./errorSamples');
 const logger = require('winston');
-const enqueue = require('../sampleQueue/sampleQueueOps').enqueue;
+const queueUtils = require('../utils/queueUtils');
 const httpStatus = require('../constants').httpStatus;
+const bulkUpsertSampleQueue = require('../constants').bulkUpsertSampleQueue;
+const commonUtils = require('../utils/commonUtils');
 
 /**
  * Validates the response from the collect function. Confirms that it is an
@@ -95,7 +97,8 @@ function handleCollectResponse(collectResponse) {
         url: ${collectRes.preparedUrl},
         numSamples: ${samplesToEnqueue.length},
       }`);
-      return enqueue(samplesToEnqueue);
+      queueUtils.enqueueFromArray(bulkUpsertSampleQueue, samplesToEnqueue,
+        commonUtils.validateSample);
     } else {
       /*
        * The transform is *not* a string, so handle the response based on the
@@ -130,7 +133,8 @@ function handleCollectResponse(collectResponse) {
           url: ${collectRes.preparedUrl},
           numSamples: ${samplesToEnqueue.length},
         }`);
-        return enqueue(samplesToEnqueue);
+        queueUtils.enqueueFromArray(bulkUpsertSampleQueue, samplesToEnqueue,
+          commonUtils.validateSample);
       } else {
         /*
          * If there is no transform designated for this HTTP status code, just
@@ -145,7 +149,8 @@ function handleCollectResponse(collectResponse) {
           error: ${errorMessage},
           numSamples: ${samplesToEnqueue.length},
         }`);
-        return enqueue(samplesToEnqueue);
+        queueUtils.enqueueFromArray(bulkUpsertSampleQueue, samplesToEnqueue,
+          commonUtils.validateSample);
       }
     }
   })

src/utils/commonUtils.js

@@ -17,6 +17,7 @@ const debug = require('debug')('refocus-collector:commonUtils');
 const path = require('path');
 const os = require('os');
 const crypto = require('crypto');
+const sampleSchema = require('./schema').sample;
 
 module.exports = {
 
@@ -75,6 +76,22 @@ module.exports = {
     return fileContents;
   },
 
+  /**
+   * Validates the sample.
+   *
+   * @param {Object} sample - The sample to validate
+   * @returns {Object} the valid sample
+   * @throws {ValidationError} if the object does not look like a sample
+   */
+  validateSample(sample) {
+    const val = sampleSchema.validate(sample);
+    if (val.error) {
+      throw new errors.ValidationError(val.error.message);
+    }
+
+    return sample;
+  },
+
   /**
    * Get current os and process metadata
    * @returns {Object} - metadata object