diff --git a/package-lock.json b/package-lock.json index 0b9bc9f7..8e9ed3bb 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,6 +1,6 @@ { "name": "@splitsoftware/splitio-commons", - "version": "0.1.1-canary.16", + "version": "0.1.1-canary.17", "lockfileVersion": 1, "requires": true, "dependencies": { diff --git a/package.json b/package.json index 35b9b864..6e44d0cb 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@splitsoftware/splitio-commons", - "version": "0.1.1-canary.16", + "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/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/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);