diff --git a/package-lock.json b/package-lock.json index e1ea1f9c..8e9ed3bb 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,6 +1,6 @@ { "name": "@splitsoftware/splitio-commons", - "version": "0.1.1-canary.15", + "version": "0.1.1-canary.17", "lockfileVersion": 1, "requires": true, "dependencies": { diff --git a/package.json b/package.json index a8341a1f..6e44d0cb 100644 --- a/package.json +++ b/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", diff --git a/src/__tests__/mocks/message.STREAMING_RESET.json b/src/__tests__/mocks/message.STREAMING_RESET.json index ee1c13ea..791d9ded 100644 --- a/src/__tests__/mocks/message.STREAMING_RESET.json +++ b/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\\\"}\"}" } \ No newline at end of file diff --git a/src/readiness/readinessManager.ts b/src/readiness/readinessManager.ts index 2c149708..394ba056 100644 --- a/src/readiness/readinessManager.ts +++ b/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); + } } } diff --git a/src/storages/AbstractSplitsCacheSync.ts b/src/storages/AbstractSplitsCacheSync.ts index 3c0c251e..9bc31467 100644 --- a/src/storages/AbstractSplitsCacheSync.ts +++ b/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; } /** diff --git a/src/sync/offline/syncTasks/fromObjectSyncTask.ts b/src/sync/offline/syncTasks/fromObjectSyncTask.ts index 0ae98763..0f1b3597 100644 --- a/src/sync/offline/syncTasks/fromObjectSyncTask.ts +++ b/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 { - 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; }); diff --git a/src/sync/streaming/SSEHandler/NotificationKeeper.ts b/src/sync/streaming/SSEHandler/NotificationKeeper.ts index ce2ba8f1..1df88250 100644 --- a/src/sync/streaming/SSEHandler/NotificationKeeper.ts +++ b/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)) { diff --git a/src/sync/streaming/SSEHandler/__tests__/index.spec.ts b/src/sync/streaming/SSEHandler/__tests__/index.spec.ts index 9324bf84..79f8c573 100644 --- a/src/sync/streaming/SSEHandler/__tests__/index.spec.ts +++ b/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 }); diff --git a/src/sync/streaming/SSEHandler/index.ts b/src/sync/streaming/SSEHandler/index.ts index de9b5eba..b0cd7069 100644 --- a/src/sync/streaming/SSEHandler/index.ts +++ b/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; } diff --git a/src/sync/streaming/SSEHandler/types.ts b/src/sync/streaming/SSEHandler/types.ts index db271614..937fb280 100644 --- a/src/sync/streaming/SSEHandler/types.ts +++ b/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 } diff --git a/src/sync/streaming/constants.ts b/src/sync/streaming/constants.ts index da381707..8afb41a6 100644 --- a/src/sync/streaming/constants.ts +++ b/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', } diff --git a/src/sync/streaming/pushManager.ts b/src/sync/streaming/pushManager.ts index 52371696..93d47127 100644 --- a/src/sync/streaming/pushManager.ts +++ b/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. diff --git a/src/sync/streaming/types.ts b/src/sync/streaming/types.ts index 8c8b59e4..6832a335 100644 --- a/src/sync/streaming/types.ts +++ b/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 MY_SEGMENTS_UPDATE ? IMySegmentsUpdateData : diff --git a/src/sync/submitters/metricsSyncTask.ts b/src/sync/submitters/metricsSyncTask.ts index 2f743884..a5264b14 100644 --- a/src/sync/submitters/metricsSyncTask.ts +++ b/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('latencies')); + return submitterSyncTaskFactory(log, postMetricsLatencies, latenciesCache, metricsRefreshRate, 'latency metrics', latencyTracker, fromCache('latencies'), 0, true); } diff --git a/src/sync/submitters/submitterSyncTask.ts b/src/sync/submitters/submitterSyncTask.ts index e0149bc8..bf92b2e9 100644 --- a/src/sync/submitters/submitterSyncTask.ts +++ b/src/sync/submitters/submitterSyncTask.ts @@ -17,6 +17,7 @@ export function submitterSyncTaskFactory( latencyTracker?: ITimeTracker, fromCacheToPayload?: (cacheData: TState) => any, maxRetries: number = 0, + debugLogs?: boolean ): ISyncTask<[], void> { let retries = 0; @@ -27,7 +28,7 @@ export function submitterSyncTaskFactory( 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); diff --git a/src/utils/settingsValidation/__tests__/splitFilters.spec.ts b/src/utils/settingsValidation/__tests__/splitFilters.spec.ts index 3aa994dd..1e23b752 100644 --- a/src/utils/settingsValidation/__tests__/splitFilters.spec.ts +++ b/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: [] } diff --git a/src/utils/settingsValidation/splitFilters.ts b/src/utils/settingsValidation/splitFilters.ts index 8a7a1303..eb7b0371 100644 --- a/src/utils/settingsValidation/splitFilters.ts +++ b/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);