[FSSDK-11003] disposable service implementation

lib/event_processor/batch_event_processor.spec.ts

@@ -167,7 +167,7 @@ describe('QueueingEventProcessor', async () => {
 
       processor.start();
       await processor.onRunning();
-      for(let i = 0; i < 100; i++) {
+      for(let i = 0; i < 99; i++) {
         const event = createImpressionEvent(`id-${i}`);
         await processor.process(event);
       }
@@ -190,30 +190,33 @@ describe('QueueingEventProcessor', async () => {
       await processor.onRunning();
 
       let events: ProcessableEvent[] = [];
-      for(let i = 0; i < 100; i++) {
+      for(let i = 0; i < 99; i++){
         const event = createImpressionEvent(`id-${i}`);
         events.push(event);
         await processor.process(event);
       }
 
       expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(0);
 
-      let event = createImpressionEvent('id-100');
+      let event = createImpressionEvent('id-99');
+      events.push(event);
       await processor.process(event);
-
+ 
       expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(1);
       expect(eventDispatcher.dispatchEvent.mock.calls[0][0]).toEqual(buildLogEvent(events));
 
-      events = [event];
-      for(let i = 101; i < 200; i++) {
+      events = [];
+
+      for(let i = 100; i < 199; i++) {
         const event = createImpressionEvent(`id-${i}`);
         events.push(event);
         await processor.process(event);
       }
 
       expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(1);
 
-      event = createImpressionEvent('id-200');
+      event = createImpressionEvent('id-199');
+      events.push(event);
       await processor.process(event);
 
       expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(2);
@@ -257,6 +260,33 @@ describe('QueueingEventProcessor', async () => {
       expect(eventDispatcher.dispatchEvent.mock.calls[1][0]).toEqual(buildLogEvent([newEvent]));
     });
 
+    it('should flush queue immediately regardless of batchSize, if event processor is disposable', async () => {
+      const eventDispatcher = getMockDispatcher();
+      const mockDispatch: MockInstance<typeof eventDispatcher.dispatchEvent> = eventDispatcher.dispatchEvent;
+      mockDispatch.mockResolvedValue({});
+
+      const dispatchRepeater = getMockRepeater();
+
+      const processor = new BatchEventProcessor({
+        eventDispatcher,
+        dispatchRepeater,
+        batchSize: 100,
+      });
+
+      processor.makeDisposable();
+      processor.start();
+      await processor.onRunning();
+
+      const events: ProcessableEvent[] = [];
+      const event = createImpressionEvent('id-1');
+      events.push(event);
+      await processor.process(event);
+
+      expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(1);
+      expect(eventDispatcher.dispatchEvent.mock.calls[0][0]).toEqual(buildLogEvent(events));
+      expect(dispatchRepeater.reset).toHaveBeenCalledTimes(1);
+    });
+
     it('should store the event in the eventStore with increasing ids', async () => {
       const eventDispatcher = getMockDispatcher();
       const eventStore = getMockSyncCache<EventWithId>();

lib/event_processor/batch_event_processor.ts

@@ -30,6 +30,9 @@ import { areEventContextsEqual } from "./event_builder/user_event";
 import { EVENT_PROCESSOR_STOPPED, FAILED_TO_DISPATCH_EVENTS, FAILED_TO_DISPATCH_EVENTS_WITH_ARG } from "../exception_messages";
 import { sprintf } from "../utils/fns";
 
+export const DEFAULT_MIN_BACKOFF = 1000;
+export const DEFAULT_MAX_BACKOFF = 32000;
+
 export type EventWithId = {
   id: string;
   event: ProcessableEvent;
@@ -209,7 +212,8 @@ export class BatchEventProcessor extends BaseService implements EventProcessor {
     if (!batch) {
       return;
     }
-
+
+    this.dispatchRepeater.reset();
     this.dispatchBatch(batch, closing);
   }
 
@@ -218,10 +222,6 @@ export class BatchEventProcessor extends BaseService implements EventProcessor {
       return Promise.reject('Event processor is not running');
     }
 
-    if (this.eventQueue.length == this.batchSize) {
-      this.flush();
-    }
-
     const eventWithId = {
       id: this.idGenerator.getId(),
       event: event,
@@ -232,13 +232,30 @@ export class BatchEventProcessor extends BaseService implements EventProcessor {
     if (this.eventQueue.length > 0 && !areEventContextsEqual(this.eventQueue[0].event, event)) {
       this.flush();
     }
-    this.eventQueue.push(eventWithId);
+
+    this.eventQueue.push(eventWithId); 
+
+    if (this.eventQueue.length == this.batchSize) {
+      this.flush();
+    } else if (!this.dispatchRepeater.isRunning()) {
+      this.dispatchRepeater.start();
+    }
+
   }
 
   start(): void {
     if (!this.isNew()) {
       return;
     }
+    if(this.disposable) {
+      this.batchSize = 1;
+      this.retryConfig = {
+        maxRetries: Math.min(this.retryConfig?.maxRetries ?? 5, 5),
+        backoffProvider:
+          this.retryConfig?.backoffProvider ||
+          (() => new ExponentialBackoff(DEFAULT_MIN_BACKOFF, DEFAULT_MAX_BACKOFF, 500)),
+      };
+    }
     super.start();
     this.state = ServiceState.Running;
     this.dispatchRepeater.start();
@@ -254,7 +271,6 @@ export class BatchEventProcessor extends BaseService implements EventProcessor {
     }
 
     if (this.isNew()) {
-      // TOOD: replace message with imported constants
       this.startPromise.reject(new Error(EVENT_PROCESSOR_STOPPED));
     }
 

lib/event_processor/event_processor_factory.spec.ts

@@ -15,8 +15,8 @@
  */
 
 import { describe, it, expect, beforeEach, vi, MockInstance } from 'vitest';
-import { DEFAULT_EVENT_BATCH_SIZE, DEFAULT_EVENT_FLUSH_INTERVAL, DEFAULT_MAX_BACKOFF, DEFAULT_MIN_BACKOFF, getBatchEventProcessor } from './event_processor_factory';
-import { BatchEventProcessor, BatchEventProcessorConfig, EventWithId } from './batch_event_processor';
+import { DEFAULT_EVENT_BATCH_SIZE, DEFAULT_EVENT_FLUSH_INTERVAL, getBatchEventProcessor } from './event_processor_factory';
+import { BatchEventProcessor, BatchEventProcessorConfig, EventWithId,DEFAULT_MAX_BACKOFF, DEFAULT_MIN_BACKOFF } from './batch_event_processor';
 import { ExponentialBackoff, IntervalRepeater } from '../utils/repeater/repeater';
 import { getMockSyncCache } from '../tests/mock/mock_cache';
 import { LogLevel } from '../modules/logging';

lib/event_processor/event_processor_factory.ts

@@ -19,14 +19,12 @@ import { StartupLog } from "../service";
 import { ExponentialBackoff, IntervalRepeater } from "../utils/repeater/repeater";
 import { EventDispatcher } from "./event_dispatcher/event_dispatcher";
 import { EventProcessor } from "./event_processor";
-import { BatchEventProcessor, EventWithId, RetryConfig } from "./batch_event_processor";
+import { BatchEventProcessor, DEFAULT_MAX_BACKOFF, DEFAULT_MIN_BACKOFF, EventWithId, RetryConfig } from "./batch_event_processor";
 import { AsyncPrefixCache, Cache, SyncPrefixCache } from "../utils/cache/cache";
 
 export const DEFAULT_EVENT_BATCH_SIZE = 10;
 export const DEFAULT_EVENT_FLUSH_INTERVAL = 1000;
 export const DEFAULT_EVENT_MAX_QUEUE_SIZE = 10000;
-export const DEFAULT_MIN_BACKOFF = 1000;
-export const DEFAULT_MAX_BACKOFF = 32000;
 export const FAILED_EVENT_RETRY_INTERVAL = 20 * 1000; 
 export const EVENT_STORE_PREFIX = 'optly_event:';
 

lib/exception_messages.ts

@@ -31,7 +31,6 @@ export const DATAFILE_MANAGER_STOPPED = 'Datafile manager stopped before it coul
 export const DATAFILE_MANAGER_FAILED_TO_START = 'Datafile manager failed to start';
 export const FAILED_TO_FETCH_DATAFILE = 'Failed to fetch datafile';
 export const FAILED_TO_STOP = 'Failed to stop';
-export const YOU_MUST_PROVIDE_DATAFILE_IN_SSR = 'You must provide datafile in SSR';
 export const YOU_MUST_PROVIDE_AT_LEAST_ONE_OF_SDKKEY_OR_DATAFILE = 'You must provide at least one of sdkKey or datafile';
 export const RETRY_CANCELLED = 'Retry cancelled';
 export const REQUEST_TIMEOUT = 'Request timeout';

lib/odp/event_manager/odp_event_manager.spec.ts

@@ -207,6 +207,39 @@ describe('DefaultOdpEventManager', () => {
     }
   });
 
+  it('should flush the queue immediately if disposable, regardless of the batchSize', async () => {
+    const apiManager = getMockApiManager();
+    const repeater = getMockRepeater()
+    apiManager.sendEvents.mockResolvedValue({ statusCode: 200 });
+    // spy on the flush method
+    const odpEventManager = new DefaultOdpEventManager({
+      repeater, 
+      apiManager: apiManager,
+      batchSize: 10,
+      retryConfig: {
+        maxRetries: 3,
+        backoffProvider: vi.fn(),
+      },
+    });
+
+    odpEventManager.updateConfig({
+      integrated: true,
+      odpConfig: config,
+    });
+    odpEventManager.makeDisposable();
+    odpEventManager.start();
+
+    await expect(odpEventManager.onRunning()).resolves.not.toThrow(); 
+
+    const event = makeEvent(0);
+    odpEventManager.sendEvent(event);
+    await exhaustMicrotasks();
+
+    expect(apiManager.sendEvents).toHaveBeenCalledTimes(1);
+    expect(apiManager.sendEvents).toHaveBeenNthCalledWith(1, config, [event]);
+    expect(repeater.reset).toHaveBeenCalledTimes(1);
+  })
+
   it('drops events and logs if the state is not running', async () => {
     const apiManager = getMockApiManager();
     apiManager.sendEvents.mockResolvedValue({ statusCode: 200 });

lib/odp/event_manager/odp_event_manager.ts

@@ -105,6 +105,11 @@ export class DefaultOdpEventManager extends BaseService implements OdpEventManag
     if (!this.isNew) {
       return;
     }
+    // Override for disposable event manager 
+    if(this.disposable) {
+      this.retryConfig.maxRetries = Math.min(this.retryConfig.maxRetries, 5);
+      this.batchSize = 1
+    }
 
     super.start();
     if (this.odpIntegrationConfig) {

lib/odp/odp_manager.spec.ts

@@ -51,6 +51,7 @@ const getMockOdpEventManager = () => {
     getState: vi.fn(),
     updateConfig: vi.fn(),
     sendEvent: vi.fn(),
+    makeDisposable: vi.fn(),
   };
 };
 

lib/odp/odp_manager.ts

@@ -90,6 +90,10 @@ export class DefaultOdpManager extends BaseService implements OdpManager {
     if (!this.isNew()) {
       return;
     }
+
+    if(this.disposable) {
+      this.eventManager.makeDisposable();
+    }
 
     this.state = ServiceState.Starting;
 

lib/optimizely/index.spec.ts

@@ -16,13 +16,10 @@
 import { describe, it, expect, vi } from 'vitest';
 import Optimizely from '.';
 import { getMockProjectConfigManager } from '../tests/mock/mock_project_config_manager';
-import * as logger from '../plugins/logger';
 import * as jsonSchemaValidator from '../utils/json_schema_validator';
-import { LOG_LEVEL } from '../common_exports';
 import { createNotificationCenter } from '../notification_center';
 import testData from '../tests/test_data';
 import { getForwardingEventProcessor } from '../event_processor/forwarding_event_processor';
-import { LoggerFacade } from '../modules/logging';
 import { createProjectConfig } from '../project_config/project_config';
 import { getMockLogger } from '../tests/mock/mock_logger';
 
@@ -39,12 +36,12 @@ describe('Optimizely', () => {
 
   const notificationCenter = createNotificationCenter({ logger, errorHandler });
 
-  it('should pass ssr to the project config manager', () => {
+  it('should pass disposable option to the project config manager', () => {
     const projectConfigManager = getMockProjectConfigManager({
       initConfig: createProjectConfig(testData.getTestProjectConfig()),
     });
-
-    vi.spyOn(projectConfigManager, 'setSsr');
+    
+    vi.spyOn(projectConfigManager, 'makeDisposable');
 
     const instance = new Optimizely({
       clientEngine: 'node-sdk',
@@ -54,16 +51,16 @@ describe('Optimizely', () => {
       logger,
       notificationCenter,
       eventProcessor,
-      isSsr: true,
+      disposable: true,
       isValidInstance: true,
     });
 
-    expect(projectConfigManager.setSsr).toHaveBeenCalledWith(true);
+    expect(projectConfigManager.makeDisposable).toHaveBeenCalled();
     // eslint-disable-next-line @typescript-eslint/ban-ts-comment
     // @ts-ignore
     expect(instance.getProjectConfig()).toBe(projectConfigManager.config);
     // eslint-disable-next-line @typescript-eslint/ban-ts-comment
     // @ts-ignore
-    expect(projectConfigManager.isSsr).toBe(true);
+    expect(projectConfigManager.disposable).toBe(true);
   });
 });

lib/optimizely/index.ts

@@ -141,10 +141,20 @@ export default class Optimizely implements Client {
     this.errorHandler = config.errorHandler;
     this.isOptimizelyConfigValid = config.isValidInstance;
     this.logger = config.logger;
+    this.projectConfigManager = config.projectConfigManager;
+    this.notificationCenter = config.notificationCenter;
     this.odpManager = config.odpManager;
     this.vuidManager = config.vuidManager;
+    this.eventProcessor = config.eventProcessor;
+
+    if(config.disposable) {
+      this.projectConfigManager.makeDisposable();
+      this.eventProcessor?.makeDisposable();
+      this.odpManager?.makeDisposable();
+    }
 
     let decideOptionsArray = config.defaultDecideOptions ?? [];
+
     if (!Array.isArray(decideOptionsArray)) {
       this.logger.log(LOG_LEVEL.DEBUG, INVALID_DEFAULT_DECIDE_OPTIONS, MODULE_NAME);
       decideOptionsArray = [];
@@ -160,7 +170,6 @@ export default class Optimizely implements Client {
       }
     });
     this.defaultDecideOptions = defaultDecideOptions;
-    this.projectConfigManager = config.projectConfigManager;
 
     this.disposeOnUpdate = this.projectConfigManager.onUpdate((configObj: projectConfig.ProjectConfig) => {
       this.logger.log(
@@ -176,7 +185,6 @@ export default class Optimizely implements Client {
       this.updateOdpSettings();
     });
 
-    this.projectConfigManager.setSsr(config.isSsr) 
     this.projectConfigManager.start();
     const projectConfigManagerRunningPromise = this.projectConfigManager.onRunning();
 
@@ -198,10 +206,6 @@ export default class Optimizely implements Client {
       UNSTABLE_conditionEvaluators: config.UNSTABLE_conditionEvaluators,
     });
 
-    this.notificationCenter = config.notificationCenter;
-
-    this.eventProcessor = config.eventProcessor;
-
     this.eventProcessor?.start();
     const eventProcessorRunningPromise = this.eventProcessor ? this.eventProcessor.onRunning() :
       Promise.resolve(undefined);
@@ -210,6 +214,8 @@ export default class Optimizely implements Client {
       this.notificationCenter.sendNotifications(NOTIFICATION_TYPES.LOG_EVENT, event);
     });
 
+    this.odpManager?.start();
+
     this.readyPromise = Promise.all([
       projectConfigManagerRunningPromise,
       eventProcessorRunningPromise,

lib/project_config/polling_datafile_manager.spec.ts

@@ -470,6 +470,25 @@ describe('PollingDatafileManager', () => {
       expect(repeater.stop).toHaveBeenCalled();
     });
 
+    it('stops repeater after successful initialization if disposable is true', async () => {
+      const repeater = getMockRepeater();
+      const requestHandler = getMockRequestHandler();
+      const mockResponse = getMockAbortableRequest(Promise.resolve({ statusCode: 200, body: '{"foo": "bar"}', headers: {} }));
+      requestHandler.makeRequest.mockReturnValueOnce(mockResponse);
+
+      const manager = new PollingDatafileManager({
+        repeater,
+        requestHandler,
+        sdkKey: 'keyThatExists',
+      });
+      manager.makeDisposable();
+      manager.start();
+      repeater.execute(0);
+
+      await expect(manager.onRunning()).resolves.not.toThrow();
+      expect(repeater.stop).toHaveBeenCalled();
+    });
+
     it('saves the datafile in cache', async () => {
       const repeater = getMockRepeater();
       const requestHandler = getMockRequestHandler();