Update SyncTask logic

CHANGES.txt

@@ -1,3 +1,7 @@
+1.4.2 (June XX, 2022)
+ - Updated telemetry to submit data even if user consent is not granted.
+ - Updated submitters logic, to avoid duplicating the post of impressions to Split cloud when the SDK is destroyed while performing its periodic post of impressions.
+
 1.4.1 (June 13, 2022)
  - Bugfixing - Updated submitters logic, to avoid dropping impressions and events that are being tracked while POST request is pending.
 

src/sync/__tests__/syncTask.spec.ts

@@ -5,7 +5,7 @@ const period = 30;
 const taskResult = 'taskResult';
 const asyncTask = jest.fn(() => Promise.resolve(taskResult));
 
-test('syncTaskFactory', (done) => {
+test('syncTaskFactory / start & stop methods for periodic execution', async () => {
 
   const syncTask = syncTaskFactory<number[], string>(loggerMock, asyncTask, period);
 
@@ -40,45 +40,75 @@ test('syncTaskFactory', (done) => {
 
   expect(syncTask.isExecuting()).toBe(true); // Executing
 
-  setTimeout(() => {
-    expect(asyncTask).toHaveBeenLastCalledWith(...startArgs); // Periodic call should be done with the initial `start` arguments
-    expect(asyncTask).toBeCalledTimes(4); // The task was executed 4 times: twice due to periodic execution and twice due to execute call
-
-    setTimeout(() => {
-      expect(asyncTask).toHaveBeenLastCalledWith(...startArgs); // Periodic call should be done with the initial `start` arguments
-      expect(asyncTask).toBeCalledTimes(5); // The task was executed 5 times: 3 due to periodic execution and twice due to execute call
-
-      // Calling `stop` stops the periodic execution of the given task
-      expect(syncTask.isRunning()).toBe(true); // Running periodically
-      syncTask.stop();
-      expect(syncTask.isRunning()).toBe(false); // Stop running periodically
-
-      setTimeout(() => {
-        expect(asyncTask).toBeCalledTimes(5); // Stopped task should not be called again
-
-        // Stopping and starting
-        syncTask.stop();
-        syncTask.start(); // Inmediatelly call task
-        syncTask.stop();
-        syncTask.start(); // Inmediatelly call task
-        syncTask.stop();
-        expect(asyncTask).toBeCalledTimes(7);
-
-        // Resume periodic execution
-        syncTask.start(); // Inmediatelly call task
-        syncTask.start(); // No effect
-        expect(asyncTask).toBeCalledTimes(8);
-
-        setTimeout(() => {
-          expect(asyncTask).toBeCalledTimes(9); // Stopped task should not be called again
-          expect(syncTask.isRunning()).toBe(true); // Running periodically
-          syncTask.stop(); // Finally stop to finish the test
-          expect(syncTask.isRunning()).toBe(false); // Stop running periodically
-
-          done();
-        }, period + 10);
-      }, period + 10);
-    }, period + 10);
-  }, period + 10);
+  await new Promise(res => setTimeout(res, period + 10));
+  expect(asyncTask).toHaveBeenLastCalledWith(...startArgs); // Periodic call should be done with the initial `start` arguments
+  expect(asyncTask).toBeCalledTimes(4); // The task was executed 4 times: twice due to periodic execution and twice due to execute call
+
+  await new Promise(res => setTimeout(res, period + 10));
+  expect(asyncTask).toHaveBeenLastCalledWith(...startArgs); // Periodic call should be done with the initial `start` arguments
+  expect(asyncTask).toBeCalledTimes(5); // The task was executed 5 times: 3 due to periodic execution and twice due to execute call
+
+  // Calling `stop` stops the periodic execution of the given task
+  expect(syncTask.isRunning()).toBe(true); // Running periodically
+  syncTask.stop();
+  expect(syncTask.isRunning()).toBe(false); // Stop running periodically
+
+  await new Promise(res => setTimeout(res, period + 10));
+  expect(asyncTask).toBeCalledTimes(5); // Stopped task should not be called again
+
+  // Stopping and starting
+  syncTask.stop();
+  syncTask.start(); // Inmediatelly call task
+  syncTask.stop();
+  syncTask.start(); // Doesn't call task since previous one has not been resolved
+  syncTask.stop();
+  expect(asyncTask).toBeCalledTimes(6);
+
+  // Resume periodic execution
+  syncTask.start(); // Doesn't call task since previous one has not been resolved
+  syncTask.start(); // No effect
+  expect(asyncTask).toBeCalledTimes(6);
+
+  await new Promise(res => setTimeout(res, period + 10));
+  expect(asyncTask).toBeCalledTimes(9); // Stopped task should not be called again
+  expect(syncTask.isRunning()).toBe(true); // Running periodically
+  syncTask.stop(); // Finally stop to finish the test
+  expect(syncTask.isRunning()).toBe(false); // Stop running periodically
+
+});
+
+test('syncTaskFactory / execute method', (done) => {
+  let executeCount = 0;
+  let resolveOrder = 0;
+  const asyncTask = jest.fn((toReturn) => {
+    executeCount++;
+    return new Promise(res => setTimeout(() => res(toReturn)));
+  });
+
+  const syncTask = syncTaskFactory(loggerMock, asyncTask, period);
+
+  syncTask.execute(1).then(result=> {
+    // console.log('1');
+    resolveOrder++;
+    expect(resolveOrder).toBe(1); // @TODO should be 1 ?
+    expect(executeCount).toBe(1);
+    expect(result).toBe(1); // @TODO should be 1
+  });
+  syncTask.execute(2).then(result=> {
+    // console.log('2');
+    resolveOrder++;
+    expect(resolveOrder).toBe(2);
+    expect(executeCount).toBe(3); // @TODO borrar
+    expect(result).toBe(2); // @TODO should be 2
+  });
+  syncTask.execute(3).then(result=> {
+    // console.log('3');
+    resolveOrder++;
+    expect(resolveOrder).toBe(3); // @TODO should be 3 ?
+    expect(executeCount).toBe(3);
+    expect(result).toBe(3);
+
+    done();
+  });
 
 });

src/sync/submitters/__tests__/eventsSubmitter.spec.ts

@@ -48,7 +48,7 @@ describe('Events submitter', () => {
     expect(eventsSubmitter.isRunning()).toEqual(false);
   });
 
-  test('without eventsFirstPushWindow', async () => {
+  test('without eventsFirstPushWindow', (done) => {
     const eventsFirstPushWindow = 0;
     params.settings.startup.eventsFirstPushWindow = eventsFirstPushWindow; // @ts-ignore
     const eventsSubmitter = eventsSubmitterFactory(params);
@@ -58,14 +58,19 @@ describe('Events submitter', () => {
     expect(eventsSubmitter.isExecuting()).toEqual(true); // and executes immediatelly if there isn't a push window
     expect(eventsCacheMock.isEmpty).toBeCalledTimes(1);
 
-    // If queue is full, submitter should be executed
+    // If queue is full, submitter is executed again after current execution is resolved
     __onFullQueueCb();
-    expect(eventsSubmitter.isExecuting()).toEqual(true);
-    expect(eventsCacheMock.isEmpty).toBeCalledTimes(2);
+    expect(eventsCacheMock.isEmpty).toBeCalledTimes(1);
 
-    expect(eventsSubmitter.isRunning()).toEqual(true);
-    eventsSubmitter.stop();
-    expect(eventsSubmitter.isRunning()).toEqual(false);
+    setTimeout(()=> {
+      expect(eventsSubmitter.isExecuting()).toEqual(false);
+      expect(eventsCacheMock.isEmpty).toBeCalledTimes(2); // 2 executions: 1st due to start and 2nd due to full queue
+
+      expect(eventsSubmitter.isRunning()).toEqual(true);
+      eventsSubmitter.stop();
+      expect(eventsSubmitter.isRunning()).toEqual(false);
+      done();
+    });
   });
 
   test('doesn\'t drop items from cache when POST is resolved', (done) => {

src/sync/submitters/__tests__/impressionsSubmitter.spec.ts

@@ -74,4 +74,26 @@ describe('Impressions submitter', () => {
     }, params.settings.scheduler.impressionsPushRate + 10);
   });
 
+  test('if it is executed while POST is pending, execution is queued until POST is resolved and not same items are submitted', (done) => {
+    // Make the POST request fail
+    params.splitApi.postTestImpressionsBulk.mockImplementation(() => Promise.resolve());
+
+    impressionsCacheInMemory.track([imp1]);
+    impressionsSubmitter.start();
+
+    // Tracking impression and executing submitter while POST is pending
+    impressionsCacheInMemory.track([{ ...imp1, keyName: 'k2' }]);
+    impressionsSubmitter.execute().then(() => {
+      expect(params.splitApi.postTestImpressionsBulk.mock.calls).toEqual([
+        // impression for k1
+        ['[{"f":"someFeature","i":[{"k":"k1","t":"someTreatment","m":0,"c":123}]}]'],
+        // impression for k2
+        ['[{"f":"someFeature","i":[{"k":"k2","t":"someTreatment","m":0,"c":123}]}]']]);
+      impressionsSubmitter.stop();
+
+      done();
+    });
+
+  });
+
 });

src/sync/syncTask.ts

@@ -4,8 +4,8 @@ import { ISyncTask } from './types';
 
 /**
  * Creates a syncTask that handles the periodic execution of a given task ("start" and "stop" methods).
- * The task can be executed once calling the "execute" method.
- * NOTE: Multiple calls to "execute" are not queued. Use "isExecuting" method to handle synchronization.
+ * The task can be also executed by calling the "execute" method. Multiple execute calls are chained to run secuentially and avoid race conditions.
+ * For example, submitters executed on SDK destroy or full queue, while periodic execution is pending.
  *
  * @param log  Logger instance.
  * @param task  Task to execute that returns a promise that NEVER REJECTS. Otherwise, periodic execution can result in Unhandled Promise Rejections.
@@ -15,8 +15,8 @@ import { ISyncTask } from './types';
  */
 export function syncTaskFactory<Input extends any[], Output = any>(log: ILogger, task: (...args: Input) => Promise<Output>, period: number, taskName = 'task'): ISyncTask<Input, Output> {
 
-  // Flag that indicates if the task is being executed
-  let executing = false;
+  // Task promise while it is pending. Undefined once the promise is resolved
+  let pendingTask: Promise<Output> | undefined;
   // flag that indicates if the task periodic execution has been started/stopped.
   let running = false;
   // Auxiliar counter used to avoid race condition when calling `start` & `stop` intermittently
@@ -26,14 +26,21 @@ export function syncTaskFactory<Input extends any[], Output = any>(log: ILogger,
   // Id of the periodic call timeout
   let timeoutID: any;
 
-  function execute(...args: Input) {
-    executing = true;
+  function execute(...args: Input): Promise<Output> {
+    // If task is executing, chain the new execution
+    if (pendingTask) {
+      return pendingTask.then(() => {
+        return execute(...args);
+      });
+    }
+
+    // Execute task
     log.debug(SYNC_TASK_EXECUTE, [taskName]);
-    return task(...args).then(result => {
-      executing = false;
+    pendingTask = task(...args).then(result => {
+      pendingTask = undefined;
       return result;
     });
-    // No need to handle promise rejection because it is a pre-condition that provided task never rejects.
+    return pendingTask;
   }
 
   function periodicExecute(currentRunningId: number) {
@@ -46,11 +53,10 @@ export function syncTaskFactory<Input extends any[], Output = any>(log: ILogger,
   }
 
   return {
-    // @TODO check if we need to queued `execute` calls, to avoid possible race conditions on submitters and updaters with streaming.
     execute,
 
     isExecuting() {
-      return executing;
+      return pendingTask !== undefined;
     },
 
     start(...args: Input) {