Localhost mode updates for SDK_READY_FROM_CACHE

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@splitsoftware/splitio-commons",
-  "version": "0.1.1-canary.15",
+  "version": "0.1.1-canary.17",
   "description": "Split Javascript SDK common components",
   "main": "cjs/index.js",
   "module": "esm/index.js",

src/__tests__/mocks/message.STREAMING_RESET.json

@@ -1,4 +1,4 @@
 {
   "type": "message",
-  "data": "{\"timestamp\":1457552660000,\"data\":\"{\\\"type\\\":\\\"STREAMING_RESET\\\"}\"}"
+  "data": "{\"timestamp\":1457552660000,\"data\":\"{\\\"type\\\":\\\"CONTROL\\\",\\\"controlType\\\":\\\"STREAMING_RESET\\\"}\"}"
 }

src/readiness/readinessManager.ts

@@ -41,7 +41,7 @@ export function readinessManagerFactory(
 
   // emit SDK_READY_FROM_CACHE
   let isReadyFromCache = false;
-  if (splits.splitsCacheLoaded) setTimeout(checkIsReadyFromCache, 0); // don't check status inmediately, to allow attach listeners
+  if (splits.splitsCacheLoaded) isReadyFromCache = true; // ready from cache, but doesn't emit SDK_READY_FROM_CACHE
   else splits.once(SDK_SPLITS_CACHE_LOADED, checkIsReadyFromCache);
 
   // emit SDK_READY_TIMED_OUT
@@ -62,13 +62,15 @@ export function readinessManagerFactory(
   let isDestroyed = false;
 
   function checkIsReadyFromCache() {
-    // @TODO add condition to emit SDK_READY_FROM_CACHE only if SDK_READY has not been emitted
-    if (!isReadyFromCache && splits.splitsCacheLoaded) {
-      // Make it async, in case user callbacks throw an exception
-      setTimeout(() => {
-        isReadyFromCache = true;
+    isReadyFromCache = true;
+    // Don't emit SDK_READY_FROM_CACHE if SDK_READY has been emitted
+    if (!isReady) {
+      try {
         gate.emit(SDK_READY_FROM_CACHE);
-      }, 0);
+      } catch (e) {
+        // throws user callback exceptions in next tick
+        setTimeout(() => { throw e; }, 0);
+      }
     }
   }
 

src/storages/AbstractSplitsCacheSync.ts

@@ -50,7 +50,7 @@ export default abstract class AbstractSplitsCacheSync implements ISplitsCacheSyn
    * It is used as condition to emit SDK_SPLITS_CACHE_LOADED, and then SDK_READY_FROM_CACHE.
    */
   checkCache(): boolean {
-    return this.getChangeNumber() > -1;
+    return false;
   }
 
   /**

src/sync/offline/syncTasks/fromObjectSyncTask.ts

@@ -7,7 +7,7 @@ import syncTaskFactory from '../../syncTask';
 import { ISyncTask } from '../../types';
 import { ISettings } from '../../../types';
 import { CONTROL } from '../../../utils/constants';
-import { SDK_SPLITS_ARRIVED, SDK_SEGMENTS_ARRIVED } from '../../../readiness/constants';
+import { SDK_SPLITS_ARRIVED, SDK_SEGMENTS_ARRIVED, SDK_SPLITS_CACHE_LOADED } from '../../../readiness/constants';
 import { SYNC_OFFLINE_DATA, ERROR_SYNC_OFFLINE_LOADING } from '../../../logger/constants';
 
 /**
@@ -20,8 +20,8 @@ export function fromObjectUpdaterFactory(
   settings: ISettings,
 ): () => Promise<boolean> {
 
-  const log = settings.log;
-  let firstTime = true;
+  const log = settings.log, splitsCache = storage.splits;
+  let startingUp = true;
 
   return function objectUpdater() {
     const splits: [string, string][] = [];
@@ -54,14 +54,20 @@ export function fromObjectUpdaterFactory(
       });
 
       return Promise.all([
-        storage.splits.clear(),
-        storage.splits.addSplits(splits)
+        splitsCache.clear(), // required to sync removed splits from mock
+        splitsCache.setChangeNumber(Date.now()),
+        splitsCache.addSplits(splits)
       ]).then(() => {
         readiness.splits.emit(SDK_SPLITS_ARRIVED);
-        // Only emits SDK_SEGMENTS_ARRIVED the first time for SDK_READY
-        if (firstTime) {
-          firstTime = false;
-          readiness.segments.emit(SDK_SEGMENTS_ARRIVED);
+
+        if (startingUp) {
+          startingUp = false;
+          Promise.resolve(splitsCache.checkCache()).then(cacheReady => {
+            // Emits SDK_READY_FROM_CACHE
+            if (cacheReady) readiness.splits.emit(SDK_SPLITS_CACHE_LOADED);
+            // Emits SDK_READY
+            readiness.segments.emit(SDK_SEGMENTS_ARRIVED);
+          });
         }
         return true;
       });

src/sync/streaming/SSEHandler/NotificationKeeper.ts

@@ -62,6 +62,14 @@ export default function notificationKeeperFactory(pushEmitter: IPushEventEmitter
     },
 
     handleControlEvent(controlType: ControlType, channel: string, timestamp: number) {
+      /* STREAMING_RESET control event is handled by PushManager directly since it doesn't require
+       * tracking timestamp and state like OCCUPANCY or CONTROL. It also ignores previous
+       * OCCUPANCY and CONTROL notifications, and whether PUSH_SUBSYSTEM_DOWN has been emitted or not */
+      if (controlType === ControlType.STREAMING_RESET) {
+        pushEmitter.emit(controlType);
+        return;
+      }
+
       for (let i = 0; i < channels.length; i++) {
         const c = channels[i];
         if (c.regex.test(channel)) {

src/sync/streaming/SSEHandler/__tests__/index.spec.ts

@@ -1,6 +1,6 @@
 // @ts-nocheck
 import SSEHandlerFactory from '..';
-import { PUSH_SUBSYSTEM_UP, PUSH_NONRETRYABLE_ERROR, PUSH_SUBSYSTEM_DOWN, PUSH_RETRYABLE_ERROR, MY_SEGMENTS_UPDATE, SEGMENT_UPDATE, SPLIT_KILL, SPLIT_UPDATE, MY_SEGMENTS_UPDATE_V2, STREAMING_RESET } from '../../constants';
+import { PUSH_SUBSYSTEM_UP, PUSH_NONRETRYABLE_ERROR, PUSH_SUBSYSTEM_DOWN, PUSH_RETRYABLE_ERROR, MY_SEGMENTS_UPDATE, SEGMENT_UPDATE, SPLIT_KILL, SPLIT_UPDATE, MY_SEGMENTS_UPDATE_V2, ControlType } from '../../constants';
 import { loggerMock } from '../../../../logger/__tests__/sdkLogger.mock';
 
 // update messages
@@ -80,7 +80,7 @@ test('`handleOpen` and `handlerMessage` for OCCUPANCY notifications (Notificatio
 
   expect(pushEmitter.emit).toHaveBeenLastCalledWith(SPLIT_UPDATE, { type: 'SPLIT_UPDATE', changeNumber: 1457552620999 }); // must handle update message if streaming on after an OCCUPANCY event
   sseHandler.handleMessage(streamingReset);
-  expect(pushEmitter.emit).toHaveBeenLastCalledWith(STREAMING_RESET); // must handle streaming reset
+  expect(pushEmitter.emit).toHaveBeenLastCalledWith(ControlType.STREAMING_RESET); // must handle streaming reset
   expect(pushEmitter.emit).toBeCalledTimes(7); // must not emit PUSH_SUBSYSTEM_UP if streaming is already on and another channel has publishers
 });
 
@@ -114,7 +114,7 @@ test('`handlerMessage` for CONTROL notifications (NotificationKeeper)', () => {
   sseHandler.handleMessage({ data: '{ "data": "{\\"type\\":\\"SPLIT_UPDATE\\",\\"changeNumber\\":1457552620999 }" }' });
   expect(pushEmitter.emit).toHaveBeenLastCalledWith(SPLIT_UPDATE, { type: 'SPLIT_UPDATE', changeNumber: 1457552620999 }); // must handle update message if streaming on after a CONTROL event
   sseHandler.handleMessage(streamingReset);
-  expect(pushEmitter.emit).toHaveBeenLastCalledWith(STREAMING_RESET); // must handle streaming reset
+  expect(pushEmitter.emit).toHaveBeenLastCalledWith(ControlType.STREAMING_RESET); // must handle streaming reset
 
   sseHandler.handleMessage(controlStreamingDisabledSec); // testing STREAMING_DISABLED with second region
   expect(pushEmitter.emit).toHaveBeenLastCalledWith(PUSH_NONRETRYABLE_ERROR); // must emit PUSH_NONRETRYABLE_ERROR if received a STREAMING_DISABLED control message
@@ -168,7 +168,7 @@ test('`handlerMessage` for update notifications (NotificationProcessor) and stre
   expect(pushEmitter.emit).toHaveBeenLastCalledWith(MY_SEGMENTS_UPDATE_V2, ...expectedParams); // must emit MY_SEGMENTS_UPDATE_V2 with the message parsed data
 
   sseHandler.handleMessage(streamingReset);
-  expect(pushEmitter.emit).toHaveBeenLastCalledWith(STREAMING_RESET); // must emit STREAMING_RESET
+  expect(pushEmitter.emit).toHaveBeenLastCalledWith(ControlType.STREAMING_RESET); // must emit STREAMING_RESET
 
 });
 

src/sync/streaming/SSEHandler/index.ts

@@ -1,6 +1,6 @@
 import { errorParser, messageParser } from './NotificationParser';
 import notificationKeeperFactory from './NotificationKeeper';
-import { PUSH_RETRYABLE_ERROR, PUSH_NONRETRYABLE_ERROR, OCCUPANCY, CONTROL, STREAMING_RESET, MY_SEGMENTS_UPDATE, MY_SEGMENTS_UPDATE_V2, SEGMENT_UPDATE, SPLIT_KILL, SPLIT_UPDATE } from '../constants';
+import { PUSH_RETRYABLE_ERROR, PUSH_NONRETRYABLE_ERROR, OCCUPANCY, CONTROL, MY_SEGMENTS_UPDATE, MY_SEGMENTS_UPDATE_V2, SEGMENT_UPDATE, SPLIT_KILL, SPLIT_UPDATE } from '../constants';
 import { IPushEventEmitter } from '../types';
 import { ISseEventHandler } from '../SSEClient/types';
 import { INotificationError, INotificationMessage } from './types';
@@ -68,7 +68,7 @@ export default function SSEHandlerFactory(log: ILogger, pushEmitter: IPushEventE
       log.debug(STREAMING_NEW_MESSAGE, [data]);
 
       // we only handle update events if streaming is up.
-      if (!notificationKeeper.isStreamingUp() && [OCCUPANCY, CONTROL, STREAMING_RESET].indexOf(parsedData.type) === -1)
+      if (!notificationKeeper.isStreamingUp() && [OCCUPANCY, CONTROL].indexOf(parsedData.type) === -1)
         return;
 
       switch (parsedData.type) {
@@ -91,13 +91,6 @@ export default function SSEHandlerFactory(log: ILogger, pushEmitter: IPushEventE
           notificationKeeper.handleControlEvent(parsedData.controlType, channel, timestamp);
           break;
 
-        /* STREAMING_RESET event is handled by PushManager directly since it doesn't require
-         tracking timestamp and state like OCCUPANCY or CONTROL. It also ignores previous
-         OCCUPANCY and CONTROL notifications, and whether PUSH_SUBSYSTEM_DOWN has been emitted or not */
-        case STREAMING_RESET:
-          pushEmitter.emit(STREAMING_RESET);
-          break;
-
         default:
           break;
       }

src/sync/streaming/SSEHandler/types.ts

@@ -1,5 +1,5 @@
 import { ControlType } from '../constants';
-import { MY_SEGMENTS_UPDATE, MY_SEGMENTS_UPDATE_V2, SEGMENT_UPDATE, SPLIT_UPDATE, SPLIT_KILL, CONTROL, OCCUPANCY, STREAMING_RESET } from '../types';
+import { MY_SEGMENTS_UPDATE, MY_SEGMENTS_UPDATE_V2, SEGMENT_UPDATE, SPLIT_UPDATE, SPLIT_KILL, CONTROL, OCCUPANCY } from '../types';
 
 export interface IMySegmentsUpdateData {
   type: MY_SEGMENTS_UPDATE,
@@ -65,11 +65,7 @@ export interface IOccupancyData {
   }
 }
 
-export interface IStreamingResetData {
-  type: STREAMING_RESET
-}
-
-export type INotificationData = IMySegmentsUpdateData | IMySegmentsUpdateV2Data | ISegmentUpdateData | ISplitUpdateData | ISplitKillData | IControlData | IOccupancyData | IStreamingResetData
+export type INotificationData = IMySegmentsUpdateData | IMySegmentsUpdateV2Data | ISegmentUpdateData | ISplitUpdateData | ISplitKillData | IControlData | IOccupancyData
 export type INotificationMessage = { parsedData: INotificationData, channel: string, timestamp: number, data: string }
 export type INotificationError = Event & { parsedData?: any, message?: string }
 

src/sync/streaming/constants.ts

@@ -35,10 +35,9 @@ export const SPLIT_UPDATE = 'SPLIT_UPDATE';
 export const CONTROL = 'CONTROL';
 export const OCCUPANCY = 'OCCUPANCY';
 
-export const STREAMING_RESET = 'STREAMING_RESET';
-
 export enum ControlType {
   STREAMING_DISABLED = 'STREAMING_DISABLED',
   STREAMING_PAUSED = 'STREAMING_PAUSED',
   STREAMING_RESUMED = 'STREAMING_RESUMED',
+  STREAMING_RESET = 'STREAMING_RESET',
 }

src/sync/streaming/pushManager.ts

@@ -15,7 +15,7 @@ import SSEClient from './SSEClient';
 import { IFetchAuth } from '../../services/types';
 import { ISettings } from '../../types';
 import { getMatching } from '../../utils/key';
-import { MY_SEGMENTS_UPDATE, MY_SEGMENTS_UPDATE_V2, PUSH_NONRETRYABLE_ERROR, PUSH_SUBSYSTEM_DOWN, SECONDS_BEFORE_EXPIRATION, SEGMENT_UPDATE, SPLIT_KILL, SPLIT_UPDATE, PUSH_RETRYABLE_ERROR, PUSH_SUBSYSTEM_UP, STREAMING_RESET } from './constants';
+import { MY_SEGMENTS_UPDATE, MY_SEGMENTS_UPDATE_V2, PUSH_NONRETRYABLE_ERROR, PUSH_SUBSYSTEM_DOWN, SECONDS_BEFORE_EXPIRATION, SEGMENT_UPDATE, SPLIT_KILL, SPLIT_UPDATE, PUSH_RETRYABLE_ERROR, PUSH_SUBSYSTEM_UP, ControlType } from './constants';
 import { IPlatform } from '../../sdkFactory/types';
 import { STREAMING_FALLBACK, STREAMING_REFRESH_TOKEN, STREAMING_CONNECTING, STREAMING_DISABLED, ERROR_STREAMING_AUTH, STREAMING_DISCONNECTING, STREAMING_RECONNECT, STREAMING_PARSING_MY_SEGMENTS_UPDATE_V2 } from '../../logger/constants';
 import { KeyList, UpdateStrategy } from './SSEHandler/types';
@@ -98,8 +98,8 @@ export default function pushManagerFactory(
     // Set token refresh 10 minutes before `expirationTime - issuedAt`
     const decodedToken = authData.decodedToken;
     const refreshTokenDelay = decodedToken.exp - decodedToken.iat - SECONDS_BEFORE_EXPIRATION;
-    // connDelay is present in AuthV2 but not in AuthV1
-    const connDelay = authData.connDelay || 0;
+    // Default connDelay of 60 secs
+    const connDelay = typeof authData.connDelay === 'number' && authData.connDelay >= 0 ? authData.connDelay : 60;
 
     log.info(STREAMING_REFRESH_TOKEN, [refreshTokenDelay, connDelay]);
 
@@ -212,7 +212,7 @@ export default function pushManagerFactory(
 
   /** STREAMING_RESET notification. Unlike a PUSH_RETRYABLE_ERROR, it doesn't emit PUSH_SUBSYSTEM_DOWN to fallback polling */
 
-  pushEmitter.on(STREAMING_RESET, function handleStreamingReset() {
+  pushEmitter.on(ControlType.STREAMING_RESET, function handleStreamingReset() {
     if (disconnected) return; // should never happen
 
     // Minimum required clean-up.

src/sync/streaming/types.ts

@@ -6,6 +6,7 @@ import { IFetchAuth } from '../../services/types';
 import { IStorageSync } from '../../storages/types';
 import { IEventEmitter, ISettings } from '../../types';
 import { IPlatform } from '../../sdkFactory/types';
+import { ControlType } from './constants';
 
 // Internal SDK events, subscribed by SyncManager and PushManager
 export type PUSH_SUBSYSTEM_UP = 'PUSH_SUBSYSTEM_UP'
@@ -24,9 +25,7 @@ export type SPLIT_UPDATE = 'SPLIT_UPDATE';
 export type CONTROL = 'CONTROL';
 export type OCCUPANCY = 'OCCUPANCY';
 
-export type STREAMING_RESET = 'STREAMING_RESET';
-
-export type IPushEvent = PUSH_SUBSYSTEM_UP | PUSH_SUBSYSTEM_DOWN | PUSH_NONRETRYABLE_ERROR | PUSH_RETRYABLE_ERROR | MY_SEGMENTS_UPDATE | MY_SEGMENTS_UPDATE_V2 | SEGMENT_UPDATE | SPLIT_UPDATE | SPLIT_KILL | STREAMING_RESET
+export type IPushEvent = PUSH_SUBSYSTEM_UP | PUSH_SUBSYSTEM_DOWN | PUSH_NONRETRYABLE_ERROR | PUSH_RETRYABLE_ERROR | MY_SEGMENTS_UPDATE | MY_SEGMENTS_UPDATE_V2 | SEGMENT_UPDATE | SPLIT_UPDATE | SPLIT_KILL | ControlType.STREAMING_RESET
 
 type IParsedData<T extends IPushEvent> =
   T extends MY_SEGMENTS_UPDATE ? IMySegmentsUpdateData :

src/sync/submitters/metricsSyncTask.ts

@@ -45,5 +45,5 @@ export function latenciesSyncTaskFactory(
 ): ISyncTask {
 
   // don't retry metrics.
-  return submitterSyncTaskFactory(log, postMetricsLatencies, latenciesCache, metricsRefreshRate, 'latency metrics', latencyTracker, fromCache<number[]>('latencies'));
+  return submitterSyncTaskFactory(log, postMetricsLatencies, latenciesCache, metricsRefreshRate, 'latency metrics', latencyTracker, fromCache<number[]>('latencies'), 0, true);
 }

src/sync/submitters/submitterSyncTask.ts

@@ -17,6 +17,7 @@ export function submitterSyncTaskFactory<TState extends { length?: number }>(
   latencyTracker?: ITimeTracker,
   fromCacheToPayload?: (cacheData: TState) => any,
   maxRetries: number = 0,
+  debugLogs?: boolean
 ): ISyncTask<[], void> {
 
   let retries = 0;
@@ -27,7 +28,7 @@ export function submitterSyncTaskFactory<TState extends { length?: number }>(
     const data = sourceCache.state();
 
     const dataCount: number | '' = typeof data.length === 'number' ? data.length : '';
-    log.info(SUBMITTERS_PUSH, [dataCount, dataName]);
+    log[debugLogs ? 'debug' : 'info'](SUBMITTERS_PUSH, [dataCount, dataName]);
     const latencyTrackerStop = latencyTracker && latencyTracker.start();
 
     const jsonPayload = JSON.stringify(fromCacheToPayload ? fromCacheToPayload(data) : data);

src/utils/settingsValidation/__tests__/splitFilters.spec.ts

@@ -11,7 +11,7 @@ import { SETTINGS_SPLITS_FILTER, ERROR_INVALID, ERROR_EMPTY_ARRAY, WARN_SPLITS_F
 
 describe('validateSplitFilters', () => {
 
-  let defaultOutput = {
+  const defaultOutput = {
     validFilters: [],
     queryString: null,
     groupedFilters: { byName: [], byPrefix: [] }
@@ -38,11 +38,11 @@ describe('validateSplitFilters', () => {
 
   test('Returns object with null queryString, if `splitFilters` contains invalid filters or contains filters with no values or invalid values', () => {
 
-    let splitFilters: any[] = [
+    const splitFilters: any[] = [
       { type: 'byName', values: [] },
       { type: 'byName', values: [] },
       { type: 'byPrefix', values: [] }];
-    let output = {
+    const output = {
       validFilters: [...splitFilters],
       queryString: null,
       groupedFilters: { byName: [], byPrefix: [] }

src/utils/settingsValidation/splitFilters.ts

@@ -107,7 +107,6 @@ export function validateSplitFilters(log: ILogger, maybeSplitFilters: any, mode:
   }
 
   // Validate filters and group their values by filter type inside `groupedFilters` object
-  // Assign the valid filters to the output of the validator by using filter function
   res.validFilters = maybeSplitFilters.filter((filter, index) => {
     if (filter && validateFilterType(filter.type) && Array.isArray(filter.values)) {
       res.groupedFilters[filter.type as SplitIO.SplitFilterType] = res.groupedFilters[filter.type as SplitIO.SplitFilterType].concat(filter.values);