[FSSDK-11003] disposable batch event processor

Author: junaed-optimizely

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/odp/event_manager/odp_event_manager.ts

@@ -107,7 +107,7 @@ export class DefaultOdpEventManager extends BaseService implements OdpEventManag
     }
     // Override for disposable event manager 
     if(this.disposable) {
-      this.retryConfig.maxRetries = 5;
+      this.retryConfig.maxRetries = Math.min(this.retryConfig.maxRetries, 5);
       this.batchSize = 1
     }
 

lib/project_config/project_config_manager.ts

@@ -29,9 +29,7 @@ import {
 } from '../exception_messages';
 
 interface ProjectConfigManagerConfig {
-  // TODO: Don't use object type
-  // eslint-disable-next-line  @typescript-eslint/ban-types
-  datafile?: string | object;
+  datafile?: string | Record<string, unknown>;
   jsonSchemaValidator?: Transformer<unknown, boolean>,
   datafileManager?: DatafileManager;
   logger?: LoggerFacade;
@@ -198,7 +196,6 @@ export class ProjectConfigManagerImpl extends BaseService implements ProjectConf
     }
 
     if (this.isNew() || this.isStarting()) {
-      // TOOD: replace message with imported constants
       this.startPromise.reject(new Error(DATAFILE_MANAGER_STOPPED));
     }