Add retry faceless (#171)

* use retry for faceless
* add await to emitter according #149

Author: Olha Virolainen

lib/actions/deleteObject.js

@@ -90,11 +90,11 @@ module.exports.process = async function process(message, configuration) {
     } else {
       if (results.length === 0) {
         this.logger.info('No objects are found');
-        return this.emit('data', messages.newEmptyMessage());
+        return messages.newEmptyMessage();
       }
       const err = new Error('More than one object found, can only delete 1');
       this.logger.error(err);
-      return this.emit('error', err);
+      throw err;
     }
   }
   this.logger.debug(`Preparing to delete a ${configuration.sobject} object...`);

lib/actions/lookupObject.js

@@ -67,13 +67,12 @@ module.exports.process = async function processAction(message, configuration) {
 
   if (!lookupValue) {
     if (allowCriteriaToBeOmitted) {
-      this.emit('data', messages.newMessageWithBody({}));
+      await this.emit('data', messages.newMessageWithBody({}));
       return;
     }
     const err = new Error('No unique criteria provided');
     this.logger.error(err);
-    this.emit('error', err);
-    return;
+    throw err;
   }
 
   const meta = await callJSForceMethod.call(this, configuration, 'describe');
@@ -88,8 +87,8 @@ module.exports.process = async function processAction(message, configuration) {
   if (lookupCache.hasKey(queryKey)) {
     this.logger.info('Cached response found!');
     const response = lookupCache.getResponse(queryKey);
-    // eslint-disable-next-line consistent-return
-    return this.emit('data', messages.newMessageWithBody(response));
+    await this.emit('data', messages.newMessageWithBody(response));
+    return;
   }
 
   // the query for the object and all its linked parent objects
@@ -107,11 +106,11 @@ module.exports.process = async function processAction(message, configuration) {
   if (records.length === 0) {
     if (allowZeroResults) {
       lookupCache.addRequestResponsePair(queryKey, {});
-      this.emit('data', messages.newMessageWithBody({}));
+      await this.emit('data', messages.newMessageWithBody({}));
     } else {
       const err = new Error('No objects found');
       this.logger.error(err);
-      this.emit('error', err);
+      throw (err);
     }
   } else if (records.length === 1) {
     try {
@@ -126,13 +125,14 @@ module.exports.process = async function processAction(message, configuration) {
 
       lookupCache.addRequestResponsePair(queryKey, records[0]);
       this.logger.debug('Emitting record');
-      this.emit('data', outputMessage);
+      await this.emit('data', outputMessage);
     } catch (err) {
-      this.emit('error', err);
+      this.logger.error('Lookup Object error occurred');
+      throw (err);
     }
   } else {
     const err = new Error('More than one object found');
     this.logger.error(err);
-    this.emit('error', err);
+    throw (err);
   }
 };

lib/helpers/attachment.js

@@ -6,7 +6,7 @@ const requestPromise = require('request-promise');
 const client = require('elasticio-rest-node')();
 
 const { callJSForceMethod } = require('./wrapper');
-const { getCredentials } = require('./oauth2Helper');
+const { getSecret } = require('../util');
 
 async function downloadFile(url, headers) {
   const optsDownload = {
@@ -71,7 +71,7 @@ exports.getAttachment = async function getAttachment(configuration, objectConten
   const binDataUrl = objectContent[binField.name];
   if (!binDataUrl) return;
 
-  const credentials = await getCredentials(emitter, configuration.secretId);
+  const { credentials } = await getSecret(emitter, configuration.secretId);
   const data = await downloadFile(credentials.undefined_params.instance_url + binDataUrl, {
     Authorization: `Bearer ${credentials.accessToken}`,
   });

lib/helpers/oauth2Helper.js

@@ -1,6 +1,6 @@
 /* eslint-disable no-await-in-loop */
 const { SalesForceClient } = require('../salesForceClient');
-const { getCredentials, refreshToken } = require('./oauth2Helper');
+const { getSecret, refreshToken } = require('../util');
 const { REFRESH_TOKEN_RETRIES } = require('../common.js').globalConsts;
 
 let client;
@@ -12,7 +12,7 @@ exports.callJSForceMethod = async function callJSForceMethod(configuration, meth
   const { secretId } = configuration;
   if (secretId) {
     this.logger.debug('Fetching credentials by secretId');
-    const credentials = await getCredentials(this, secretId);
+    const { credentials } = await getSecret(this, secretId);
     accessToken = credentials.access_token;
     instanceUrl = credentials.undefined_params.instance_url;
   } else {

lib/helpers/wrapper.js

@@ -168,8 +168,8 @@ class SalesForceClient {
       .on('end', () => {
         this.logger.debug('Found %s records', results.length);
       })
-      .on('error', (err) => {
-        this.emit('error', err);
+      .on('error', async (err) => {
+        await this.emit('error', err);
       })
       .execute({ autoFetch: true, maxFetch });
     return results;
@@ -222,23 +222,29 @@ class SalesForceClient {
     const sobject = options.sobject || this.configuration.sobject;
     const includeDeleted = options.includeDeleted || this.configuration.includeDeleted;
     const { wherePart, offset, limit } = options;
-    await this.connection.sobject(sobject)
-      .select('*')
-      .where(wherePart)
-      .offset(offset)
-      .limit(limit)
-      .scanAll(includeDeleted)
-      .on('error', (err) => {
-        this.logger.error('Salesforce returned an error');
-        throw err;
-      })
-      .on('record', (record) => {
-        records.push(record);
-      })
-      .on('end', () => {
-        this.logger.debug('Found %s records', records.length);
-      })
-      .execute({ autoFetch: true, maxFetch: limit });
+    try {
+      await this.connection.sobject(sobject)
+        .select('*')
+        .where(wherePart)
+        .offset(offset)
+        .limit(limit)
+        .scanAll(includeDeleted)
+        .on('error', (err) => {
+          this.logger.error('Salesforce returned an error');
+          throw err;
+        })
+        .on('record', (record) => {
+          records.push(record);
+        })
+        .on('end', () => {
+          this.logger.debug('Found %s records', records.length);
+        })
+        .execute({ autoFetch: true, maxFetch: limit });
+    } catch (e) {
+      this.logger.trace('Lookup query failed', e);
+      this.logger.error('Lookup query failed');
+      throw e;
+    }
     return records;
   }
 

lib/salesForceClient.js

@@ -1,9 +1,10 @@
 const jsforce = require('jsforce');
 const { messages } = require('elasticio-node');
 const { callJSForceMethod } = require('../helpers/wrapper');
-const { getCredentials, refreshToken } = require('../helpers/oauth2Helper');
-const { REFRESH_TOKEN_RETRIES, SALESFORCE_API_VERSION } = require('../common.js').globalConsts;
+const { getSecret, refreshToken } = require('../util');
+const { SALESFORCE_API_VERSION } = require('../common.js').globalConsts;
 
+let fayeClient;
 /**
  * This method will be called from elastic.io platform providing following data
  *
@@ -12,15 +13,13 @@ const { REFRESH_TOKEN_RETRIES, SALESFORCE_API_VERSION } = require('../common.js'
  */
 async function processTrigger(msg, configuration) {
   this.logger.info('Starting Subscribe to platform events Trigger');
-  let iteration = REFRESH_TOKEN_RETRIES;
-  let refreshTokenSuccess = false;
   const { secretId } = configuration;
   if (!secretId) {
     this.logger.error('secretId is missing in configuration, credentials cannot be fetched');
     throw new Error('secretId is missing in configuration, credentials cannot be fetched');
   }
   this.logger.debug('Fetching credentials by secretId');
-  const credentials = await getCredentials(this, secretId);
+  const { credentials } = await getSecret(this, secretId);
   const accessToken = credentials.access_token;
   const instanceUrl = credentials.undefined_params.instance_url;
   this.logger.trace('AccessToken = %s', accessToken);
@@ -33,52 +32,46 @@ async function processTrigger(msg, configuration) {
   const topic = `/event/${configuration.object}`;
   const replayId = -1;
   this.logger.debug('Creating streaming client');
-  const fayeClient = connection.streaming.createClient([
-    new jsforce.StreamingExtension.Replay(topic, replayId),
-    new jsforce.StreamingExtension.AuthFailure(async (err) => {
-      this.logger.trace('AuthFailure: %j', err);
-      // eslint-disable-next-line max-len
-      // if (err.ext && err.ext.sfdc && err.ext.sfdc.failureReason && (err.ext.sfdc.failureReason === '401::Authentication invalid') && (iteration > 0)) {
-      this.logger.trace('err.ext.sfdc.failureReason: %s', err.ext.sfdc.failureReason);
-      if (err.ext.sfdc.failureReason === '401::Authentication invalid') {
-        do {
-          iteration -= 1;
+  if (!fayeClient) {
+    fayeClient = connection.streaming.createClient([
+      new jsforce.StreamingExtension.Replay(topic, replayId),
+      new jsforce.StreamingExtension.AuthFailure(async (err) => {
+        this.logger.trace('AuthFailure: %j', err);
+        if (err.ext && err.ext.sfdc && err.ext.sfdc.failureReason && (err.ext.sfdc.failureReason === '401::Authentication invalid')) {
           try {
-            this.logger.debug('Session is expired, trying to refresh token, iteration: %s', REFRESH_TOKEN_RETRIES - iteration);
-            refreshTokenSuccess = await refreshToken(this, secretId);
-            this.logger.debug('Token is successfully refreshed, iteration: %s', REFRESH_TOKEN_RETRIES - iteration);
-            break;
+            this.logger.debug('Session is expired, trying to refresh token');
+            await refreshToken(this, secretId);
+            this.logger.debug('Token is successfully refreshed');
           } catch (error) {
-            this.logger.error('Failed to refresh token, iteration: %s', REFRESH_TOKEN_RETRIES - iteration);
+            this.logger.trace('Refresh token error: %j', error);
+            this.logger.error('Failed to fetch and/or refresh token');
+            throw new Error('Failed to fetch and/or refresh token');
           }
-        } while (iteration > 0);
-        if (!refreshTokenSuccess) {
-          this.logger.error('Failed to fetch and/or refresh token, retries exceeded');
-          throw new Error('Failed to fetch and/or refresh token, retries exceeded');
+          fayeClient = undefined;
+          this.logger.info('Lets call processTrigger one more time');
+          await processTrigger.call(this, msg, configuration);
+        } else {
+          this.logger.error('AuthFailure extension error occurred');
+          throw err;
         }
-        this.logger.info('Lets call processTrigger one more time');
-        await processTrigger.call(this, msg, configuration);
-      } else {
-        this.logger.error('AuthFailure extension error occurred');
-        throw err;
-      }
-    }),
-  ]);
+      }),
+    ]);
 
-  fayeClient.subscribe(topic, async (message) => {
-    this.logger.info('Incoming message found, going to emit...');
-    this.logger.trace('Incoming Message: %j', message);
-    await this.emit('data', messages.newMessageWithBody(message));
-  })
-    .then(() => {
-      this.logger.info('Subscribed to PushTopic successfully');
-      this.logger.trace(`Subscribed to PushTopic: ${topic}`);
-    },
-    (err) => {
-      this.logger.error('Subscriber error occurred');
-      throw err;
-    });
-  this.logger.info('Streaming client created and ready');
+    fayeClient.subscribe(topic, async (message) => {
+      this.logger.info('Incoming message found, going to emit...');
+      this.logger.trace('Incoming Message: %j', message);
+      await this.emit('data', messages.newMessageWithBody(message));
+    })
+      .then(() => {
+        this.logger.info('Subscribed to PushTopic successfully');
+        this.logger.trace(`Subscribed to PushTopic: ${topic}`);
+      },
+      (err) => {
+        this.logger.error('Subscriber error occurred');
+        throw err;
+      });
+    this.logger.info('Streaming client created and ready');
+  }
 }
 
 /**