Update PushManager to solve some corner case issues

.github/workflows/ci.yml

@@ -23,7 +23,7 @@ jobs:
       - name: Set up nodejs
         uses: actions/setup-node@v2
         with:
-          node-version: 'lts/*'
+          node-version: '14'
           cache: 'npm'
 
       - name: npm CI

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@splitsoftware/splitio-commons",
-  "version": "1.2.0",
+  "version": "1.2.1-rc.0",
   "description": "Split Javascript SDK common components",
   "main": "cjs/index.js",
   "module": "esm/index.js",

src/evaluator/matchers/ew.ts

@@ -1,13 +1,13 @@
 import { ENGINE_MATCHER_ENDS_WITH } from '../../logger/constants';
 import { ILogger } from '../../logger/types';
-import { endsWith as strEndsWith } from '../../utils/lang';
+import { endsWith } from '../../utils/lang';
 
 export function endsWithMatcherContext(log: ILogger, ruleAttr: string[]) /*: Function */ {
   return function endsWithMatcher(runtimeAttr: string): boolean {
-    let endsWith = ruleAttr.some(e => strEndsWith(runtimeAttr, e));
+    let strEndsWith = ruleAttr.some(e => endsWith(runtimeAttr, e));
 
-    log.debug(ENGINE_MATCHER_ENDS_WITH, [runtimeAttr, ruleAttr, endsWith]);
+    log.debug(ENGINE_MATCHER_ENDS_WITH, [runtimeAttr, ruleAttr, strEndsWith]);
 
-    return endsWith;
+    return strEndsWith;
   };
 }

src/logger/messages/info.ts

@@ -23,10 +23,10 @@ export const codesInfo: [number, string][] = codesWarn.concat([
   [c.SUBMITTERS_PUSH_FULL_EVENTS_QUEUE, c.LOG_PREFIX_SYNC_SUBMITTERS + 'Flushing full events queue and reseting timer.'],
   [c.SUBMITTERS_PUSH, c.LOG_PREFIX_SYNC_SUBMITTERS + 'Pushing %s %s.'],
   [c.STREAMING_REFRESH_TOKEN, c.LOG_PREFIX_SYNC_STREAMING + 'Refreshing streaming token in %s seconds, and connecting streaming in %s seconds.'],
-  [c.STREAMING_RECONNECT, c.LOG_PREFIX_SYNC_STREAMING + 'Attempting to reconnect in %s seconds.'],
-  [c.STREAMING_CONNECTING, c.LOG_PREFIX_SYNC_STREAMING + 'Connecting to streaming.'],
+  [c.STREAMING_RECONNECT, c.LOG_PREFIX_SYNC_STREAMING + 'Attempting to reconnect streaming in %s seconds.'],
+  [c.STREAMING_CONNECTING, c.LOG_PREFIX_SYNC_STREAMING + 'Connecting streaming.'],
   [c.STREAMING_DISABLED, c.LOG_PREFIX_SYNC_STREAMING + 'Streaming is disabled for given Api key. Switching to polling mode.'],
-  [c.STREAMING_DISCONNECTING, c.LOG_PREFIX_SYNC_STREAMING + 'Disconnecting from streaming.'],
+  [c.STREAMING_DISCONNECTING, c.LOG_PREFIX_SYNC_STREAMING + 'Disconnecting streaming.'],
   [c.SYNC_START_POLLING, c.LOG_PREFIX_SYNC_MANAGER + 'Streaming not available. Starting polling.'],
   [c.SYNC_CONTINUE_POLLING, c.LOG_PREFIX_SYNC_MANAGER + 'Streaming couldn\'t connect. Continue polling.'],
   [c.SYNC_STOP_POLLING, c.LOG_PREFIX_SYNC_MANAGER + 'Streaming (re)connected. Syncing and stopping polling.'],

src/sync/__tests__/syncTask.mock.ts

@@ -0,0 +1,9 @@
+export function syncTaskFactory() {
+  return {
+    execute: jest.fn(),
+    isExecuting: jest.fn(),
+    start: jest.fn(),
+    stop: jest.fn(),
+    isRunning: jest.fn(),
+  };
+}

src/sync/__tests__/syncTask.spec.ts

@@ -64,7 +64,7 @@ test('syncTaskFactory', (done) => {
         syncTask.stop();
         expect(asyncTask).toBeCalledTimes(7);
 
-        // // Resume periodic execution
+        // Resume periodic execution
         syncTask.start(); // Inmediatelly call task
         syncTask.start(); // No effect
         expect(asyncTask).toBeCalledTimes(8);

src/sync/polling/fetchers/mySegmentsFetcher.ts

@@ -6,9 +6,10 @@ import { IMySegmentsFetcher } from './types';
  * Factory of MySegments fetcher.
  * MySegments fetcher is a wrapper around `mySegments` API service that parses the response and handle errors.
  */
-export function mySegmentsFetcherFactory(fetchMySegments: IFetchMySegments, userMatchingKey: string): IMySegmentsFetcher {
+export function mySegmentsFetcherFactory(fetchMySegments: IFetchMySegments): IMySegmentsFetcher {
 
   return function mySegmentsFetcher(
+    userMatchingKey: string,
     noCache?: boolean,
     // Optional decorator for `fetchMySegments` promise, such as timeout or time tracker
     decorator?: (promise: Promise<IResponse>) => Promise<IResponse>

src/sync/polling/fetchers/types.ts

@@ -15,6 +15,7 @@ export type ISegmentChangesFetcher = (
 ) => Promise<ISegmentChangesResponse[]>
 
 export type IMySegmentsFetcher = (
+  userMatchingKey: string,
   noCache?: boolean,
   decorator?: (promise: Promise<IResponse>) => Promise<IResponse>
 ) => Promise<string[]>

src/sync/polling/pollingManagerCS.ts

@@ -1,26 +1,23 @@
 import { ISegmentsSyncTask, ISplitsSyncTask, IPollingManagerCS } from './types';
 import { forOwn } from '../../utils/lang';
 import { IReadinessManager } from '../../readiness/types';
-import { ISplitApi } from '../../services/types';
 import { IStorageSync } from '../../storages/types';
 import { mySegmentsSyncTaskFactory } from './syncTasks/mySegmentsSyncTask';
 import { splitsSyncTaskFactory } from './syncTasks/splitsSyncTask';
-import { ISettings } from '../../types';
 import { getMatching } from '../../utils/key';
 import { SDK_SPLITS_ARRIVED, SDK_SEGMENTS_ARRIVED } from '../../readiness/constants';
 import { POLLING_SMART_PAUSING, POLLING_START, POLLING_STOP } from '../../logger/constants';
+import { ISyncManagerFactoryParams } from '../types';
 
 /**
  * Expose start / stop mechanism for polling data from services.
  * For client-side API with multiple clients.
  */
 export function pollingManagerCSFactory(
-  splitApi: ISplitApi,
-  storage: IStorageSync,
-  readiness: IReadinessManager,
-  settings: ISettings,
+  params: ISyncManagerFactoryParams
 ): IPollingManagerCS {
 
+  const { splitApi, storage, readiness, settings } = params;
   const log = settings.log;
 
   const splitsSyncTask: ISplitsSyncTask = splitsSyncTaskFactory(splitApi.fetchSplitChanges, storage, readiness, settings);

src/sync/polling/pollingManagerSS.ts

@@ -1,23 +1,18 @@
 import { splitsSyncTaskFactory } from './syncTasks/splitsSyncTask';
 import { segmentsSyncTaskFactory } from './syncTasks/segmentsSyncTask';
-import { IStorageSync } from '../../storages/types';
-import { IReadinessManager } from '../../readiness/types';
-import { ISplitApi } from '../../services/types';
-import { ISettings } from '../../types';
 import { IPollingManager, ISegmentsSyncTask, ISplitsSyncTask } from './types';
 import { thenable } from '../../utils/promise/thenable';
 import { POLLING_START, POLLING_STOP, LOG_PREFIX_SYNC_POLLING } from '../../logger/constants';
+import { ISyncManagerFactoryParams } from '../types';
 
 /**
  * Expose start / stop mechanism for pulling data from services.
  */
 export function pollingManagerSSFactory(
-  splitApi: ISplitApi,
-  storage: IStorageSync,
-  readiness: IReadinessManager,
-  settings: ISettings
+  params: ISyncManagerFactoryParams
 ): IPollingManager {
 
+  const { splitApi, storage, readiness, settings } = params;
   const log = settings.log;
 
   const splitsSyncTask: ISplitsSyncTask = splitsSyncTaskFactory(splitApi.fetchSplitChanges, storage, readiness, settings);

src/sync/polling/syncTasks/mySegmentsSyncTask.ts

@@ -21,12 +21,13 @@ export function mySegmentsSyncTaskFactory(
     settings.log,
     mySegmentsUpdaterFactory(
       settings.log,
-      mySegmentsFetcherFactory(fetchMySegments, matchingKey),
+      mySegmentsFetcherFactory(fetchMySegments),
       storage.splits,
       storage.segments,
       readiness.segments,
       settings.startup.requestTimeoutBeforeReady,
       settings.startup.retriesOnFailureBeforeReady,
+      matchingKey
     ),
     settings.scheduler.segmentsRefreshRate,
     'mySegmentsUpdater',

src/sync/polling/types.ts

@@ -1,7 +1,5 @@
 import { IReadinessManager } from '../../readiness/types';
-import { ISplitApi } from '../../services/types';
 import { IStorageSync } from '../../storages/types';
-import { ISettings } from '../../types';
 import { SegmentsData } from '../streaming/SSEHandler/types';
 import { ITask, ISyncTask } from '../types';
 
@@ -23,13 +21,3 @@ export interface IPollingManagerCS extends IPollingManager {
   remove(matchingKey: string): void;
   get(matchingKey: string): ISegmentsSyncTask | undefined
 }
-
-/**
- * Signature of polling manager factory/constructor
- */
-export type IPollingManagerFactoryParams = [
-  splitApi: ISplitApi,
-  storage: IStorageSync,
-  readiness: IReadinessManager,
-  settings: ISettings,
-]

src/sync/polling/updaters/mySegmentsUpdater.ts

@@ -23,6 +23,7 @@ export function mySegmentsUpdaterFactory(
   segmentsEventEmitter: ISegmentsEventEmitter,
   requestTimeoutBeforeReady: number,
   retriesOnFailureBeforeReady: number,
+  matchingKey: string
 ): IMySegmentsUpdater {
 
   let readyOnAlreadyExistentState = true;
@@ -69,7 +70,7 @@ export function mySegmentsUpdaterFactory(
       // If segmentsData is provided, there is no need to fetch mySegments
       new Promise((res) => { updateSegments(segmentsData); res(true); }) :
       // If not provided, fetch mySegments
-      mySegmentsFetcher(noCache, _promiseDecorator).then(segments => {
+      mySegmentsFetcher(matchingKey, noCache, _promiseDecorator).then(segments => {
         // Only when we have downloaded segments completely, we should not keep retrying anymore
         startingUp = false;
 

src/sync/streaming/AuthClient/index.ts

@@ -17,8 +17,7 @@ export function authenticateFactory(fetchAuth: IFetchAuth): IAuthenticate {
    * @param {string[] | undefined} userKeys set of user Keys to track MY_SEGMENTS_CHANGES. It is undefined for server-side API.
    */
   return function authenticate(userKeys?: string[]): Promise<IAuthToken> {
-    let authPromise = fetchAuth(userKeys); // errors handled by fetchAuth service
-    return authPromise
+    return fetchAuth(userKeys)
       .then(resp => resp.json())
       .then(json => {
         if (json.token) { // empty token when `"pushEnabled": false`

src/sync/streaming/UpdateWorkers/SegmentsUpdateWorker.ts

@@ -19,7 +19,7 @@ export class SegmentsUpdateWorker implements IUpdateWorker {
    * @param {Object} segmentsCache segments data cache
    * @param {Object} segmentsSyncTask task for syncing segments data
    */
-  constructor(segmentsCache: ISegmentsCacheSync, segmentsSyncTask: ISegmentsSyncTask) {
+  constructor(segmentsSyncTask: ISegmentsSyncTask, segmentsCache: ISegmentsCacheSync) {
     this.segmentsCache = segmentsCache;
     this.segmentsSyncTask = segmentsSyncTask;
     this.maxChangeNumbers = {};

src/sync/streaming/UpdateWorkers/__tests__/SegmentsUpdateWorker.spec.ts

@@ -47,7 +47,7 @@ describe('SegmentsUpdateWorker ', () => {
     cache.addToSegment('mocked_segment_3', ['e']);
     const segmentsSyncTask = segmentsSyncTaskMock(cache);
 
-    const segmentsUpdateWorker = new SegmentsUpdateWorker(cache, segmentsSyncTask);
+    const segmentsUpdateWorker = new SegmentsUpdateWorker(segmentsSyncTask, cache);
     segmentsUpdateWorker.backoff.baseMillis = 0; // retry immediately
 
     expect(segmentsUpdateWorker.maxChangeNumbers).toEqual({}); // inits with not queued events;