streamPlatformEvents.js use platform secrets (#169)

* streamPlatformEvents.js use platform secrets
* Add retry faceless (#171)
* add await to emitter according #149

Author: Olha Virolainen

README.md

@@ -120,11 +120,6 @@ This trigger will subscribe for any platform Event using Salesforce streaming AP
 ![Screenshot from 2019-03-11 11-51-10](https://user-images.githubusercontent.com/13310949/54114889-1088e900-43f4-11e9-8b49-3a8113b6577d.png)
 
 You can find more detail information in the [Platform Events Intro Documentation](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_intro.htm).
-#### Environment Variables
-
-1. `SALESFORCE_API_VERSION` - API version for not deprecated actions and triggers e.g(46.0), default value 45.0
-
-2. `LOG_LEVEL` - `trace` | `debug` | `info` | `warning` | `error` controls logger level
 
 #### Limitations:
 At the moment this trigger can be used only for **"Realtime"** flows.

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}`,
   });