80 use replay (#81)

Author: if0s

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 2.8.2 (February 02, 2023)
+* Fixed bug when component didn't use `replayId` after error in `Subscribe to PubSub` trigger
+
 ## 2.8.1 (December 28, 2023)
 * Fixed duplicate retries and added exponential backoff in `Subscribe to PubSub` trigger
 

component.json

@@ -3,7 +3,7 @@
   "description": "Customer relationship management (CRM) software & cloud computing from the leader in CRM solutions for businesses large & small.",
   "docsUrl": "https://github.com/elasticio/salesforce-component-v2",
   "url": "http://www.salesforce.com/",
-  "version": "2.8.1",
+  "version": "2.8.2",
   "authClientTypes": [
     "oauth2"
   ],

lib/PubSubClient.js

@@ -131,6 +131,7 @@ class PubSubClient {
     const writeOptions = { topicName: this.topicName, numRequested };
 
     if (fromReplayId) {
+      this.logger.info(`Subscription will start from replayId: ${fromReplayId}`);
       const buf = Buffer.allocUnsafe(8);
       buf.writeBigUInt64BE(BigInt(fromReplayId), 0);
       writeOptions.replayPreset = 2;

lib/triggers/streamPlatformEventsPubSub.js

@@ -16,6 +16,16 @@ const MAX_AUTH_RETRIES = 2;
 const MAX_CONNECT_RETRIES = 5;
 let eventEmitter;
 let creationInProgress = false;
+let timeoutRef;
+
+const saveSnapshot = () => {
+  if (timeoutRef) clearTimeout(timeoutRef);
+  timeoutRef = setTimeout(async () => {
+    context.logger.debug(`Saving snapshot with nextStartReplayId: ${nextStartReplayId}`);
+    await context.emit('snapshot', { nextStartReplayId });
+    context.logger.debug('Snapshot saved');
+  }, 2000);
+};
 
 const reconnectHandler = async (reconnectSleepTime, cfg) => {
   context.logger.warn(`Restarting in ${Math.round(reconnectSleepTime / 1000)} seconds`);
@@ -92,7 +102,7 @@ async function processTrigger(msg, cfg, snapshot) {
   const { credentials } = await getSecret(this, secretId, msg.id);
   const accessToken = credentials.access_token;
   const instanceUrl = credentials.undefined_params.instance_url;
-  nextStartReplayId = snapshot?.nextStartReplayId || initialReplayId;
+  nextStartReplayId = nextStartReplayId || snapshot?.nextStartReplayId || initialReplayId;
 
   let emittedOrAlive = false;
   const timeOut = async (t) => {
@@ -126,18 +136,18 @@ async function processTrigger(msg, cfg, snapshot) {
   }
   context.logger.info('SalesForce connected');
 
-  eventEmitter = await client.subscribe(eventCountPerRequest, snapshot?.nextStartReplayId || initialReplayId);
+  eventEmitter = await client.subscribe(eventCountPerRequest, nextStartReplayId);
   context.logger.info('Streaming client created');
 
   eventEmitter.on('data', async (event) => {
-    context.logger.info('Got new message, emitting');
+    context.logger.info(`Got new message, replayId: ${event.event.replayId}, emitting`);
     await context.emit('data', messages.newMessageWithBody(event));
-    nextStartReplayId = event.replayId;
+    nextStartReplayId = event.event.replayId;
     emittedOrAlive = true;
-    await this.emit('snapshot', { nextStartReplayId });
+    saveSnapshot();
   });
 
-  eventEmitter.once('keepalive', (data) => { context.logger.info('Got keepalive message'); });
+  eventEmitter.once('keepalive', (data) => { context.logger.info(`Got keepalive message, replayId: ${data.latestReplayId}`); });
 
   try {
     await Promise.race([
@@ -147,7 +157,7 @@ async function processTrigger(msg, cfg, snapshot) {
           keepaliveFailureRetry = 1;
           emittedOrAlive = true;
           nextStartReplayId = data.latestReplayId;
-          await this.emit('snapshot', { nextStartReplayId });
+          saveSnapshot();
           resolve();
         });
         eventEmitter.on('error', (error) => {