feat: Add public API for setting event dispatcher (#139)

* feat: Add public API for setting event dispatcher

* nit

* test

* Make BoundedEventQueue implement NamedEventQueue

* simplify

* export BoundedEventQueue

* cleanup import

* add unused imports lint

Author: felipecsl

.eslintrc.js

@@ -12,7 +12,7 @@ module.exports = {
     'plugin:promise/recommended',
     'plugin:import/recommended',
   ],
-  plugins: ['@typescript-eslint', 'prettier', 'import', 'promise'],
+  plugins: ['@typescript-eslint', 'prettier', 'import', 'promise', 'unused-imports'],
   rules: {
     'prettier/prettier': [
       'warn',
@@ -24,6 +24,7 @@ module.exports = {
     ],
     'import/named': 'off',
     'import/no-unresolved': 'off',
+    'unused-imports/no-unused-imports': 'error',
     '@typescript-eslint/no-explicit-any': 'off',
     'import/order': [
       'warn',

package.json

@@ -55,6 +55,7 @@
     "eslint-plugin-import": "^2.25.4",
     "eslint-plugin-prettier": "^4.0.0",
     "eslint-plugin-promise": "^6.0.0",
+    "eslint-plugin-unused-imports": "^4.1.4",
     "jest": "^29.7.0",
     "jest-environment-jsdom": "^29.7.0",
     "lodash": "^4.17.21",

src/client/eppo-client.ts

@@ -20,7 +20,6 @@ import {
 import { decodeFlag } from '../decoding';
 import { EppoValue } from '../eppo_value';
 import { Evaluator, FlagEvaluation, noneResult } from '../evaluator';
-import ArrayBackedNamedEventQueue from '../events/array-backed-named-event-queue';
 import { BoundedEventQueue } from '../events/bounded-event-queue';
 import EventDispatcher from '../events/event-dispatcher';
 import NoOpEventDispatcher from '../events/no-op-event-dispatcher';
@@ -79,21 +78,12 @@ export interface IContainerExperiment<T> {
   treatmentVariationEntries: Array<T>;
 }
 
-const DEFAULT_EVENT_DISPATCHER_CONFIG = {
-  // TODO: Replace with actual ingestion URL
-  ingestionUrl: 'https://example.com/events',
-  batchSize: 10,
-  flushIntervalMs: 10_000,
-  retryIntervalMs: 5_000,
-  maxRetries: 3,
-};
-
 export default class EppoClient {
-  private readonly eventDispatcher: EventDispatcher;
+  private eventDispatcher: EventDispatcher;
   private readonly assignmentEventsQueue: BoundedEventQueue<IAssignmentEvent> =
-    newBoundedArrayEventQueue<IAssignmentEvent>('assignments');
+    new BoundedEventQueue<IAssignmentEvent>('assignments');
   private readonly banditEventsQueue: BoundedEventQueue<IBanditEvent> =
-    newBoundedArrayEventQueue<IBanditEvent>('bandit');
+    new BoundedEventQueue<IBanditEvent>('bandit');
   private readonly banditEvaluator = new BanditEvaluator();
   private banditLogger?: IBanditLogger;
   private banditAssignmentCache?: AssignmentCache;
@@ -152,6 +142,12 @@ export default class EppoClient {
     this.banditVariationConfigurationStore = banditVariationConfigurationStore;
   }
 
+  /** Sets the EventDispatcher instance to use when tracking events with {@link track}. */
+  // noinspection JSUnusedGlobalSymbols
+  setEventDispatcher(eventDispatcher: EventDispatcher) {
+    this.eventDispatcher = eventDispatcher;
+  }
+
   // noinspection JSUnusedGlobalSymbols
   setBanditModelConfigurationStore(
     banditModelConfigurationStore: IConfigurationStore<BanditParameters>,
@@ -1145,7 +1141,3 @@ export function checkValueTypeMatch(
       return false;
   }
 }
-
-function newBoundedArrayEventQueue<T>(name: string): BoundedEventQueue<T> {
-  return new BoundedEventQueue<T>(new ArrayBackedNamedEventQueue<T>(name));
-}

src/events/bounded-event-queue.ts

@@ -4,17 +4,32 @@ import { MAX_EVENT_QUEUE_SIZE } from '../constants';
 import NamedEventQueue from './named-event-queue';
 
 /** A bounded event queue that drops events when it reaches its maximum size. */
-export class BoundedEventQueue<T> {
+export class BoundedEventQueue<T> implements NamedEventQueue<T> {
   constructor(
-    private readonly queue: NamedEventQueue<T>,
+    readonly name: string,
+    private readonly queue = new Array<T>(),
     private readonly maxSize = MAX_EVENT_QUEUE_SIZE,
   ) {}
 
+  length = this.queue.length;
+
+  splice(count: number): T[] {
+    return this.queue.splice(count);
+  }
+
+  isEmpty(): boolean {
+    return this.queue.length === 0;
+  }
+
+  [Symbol.iterator](): IterableIterator<T> {
+    return this.queue[Symbol.iterator]();
+  }
+
   push(event: T) {
     if (this.queue.length < this.maxSize) {
       this.queue.push(event);
     } else {
-      logger.warn(`Dropping event for queue ${this.queue.name} since the queue is full`);
+      logger.warn(`Dropping event for queue ${this.name} since the queue is full`);
     }
   }
 

src/events/default-event-dispatcher.spec.ts

@@ -1,8 +1,9 @@
-import { resolve } from 'eslint-import-resolver-typescript';
-
 import ArrayBackedNamedEventQueue from './array-backed-named-event-queue';
 import BatchEventProcessor from './batch-event-processor';
-import DefaultEventDispatcher, { EventDispatcherConfig } from './default-event-dispatcher';
+import DefaultEventDispatcher, {
+  EventDispatcherConfig,
+  newDefaultEventDispatcher,
+} from './default-event-dispatcher';
 import { Event } from './event-dispatcher';
 import NetworkStatusListener from './network-status-listener';
 
@@ -198,4 +199,26 @@ describe('DefaultEventDispatcher', () => {
       expect(global.fetch).toHaveBeenCalled();
     });
   });
+
+  describe('newDefaultEventDispatcher', () => {
+    it('should throw if SDK key is invalid', () => {
+      expect(() => {
+        newDefaultEventDispatcher(
+          new ArrayBackedNamedEventQueue('test-queue'),
+          mockNetworkStatusListener,
+          'invalid-sdk-key',
+        );
+      }).toThrow('Unable to parse Event ingestion URL from SDK key');
+    });
+
+    it('should create a new DefaultEventDispatcher with the provided configuration', () => {
+      const eventQueue = new ArrayBackedNamedEventQueue('test-queue');
+      const dispatcher = newDefaultEventDispatcher(
+        eventQueue,
+        mockNetworkStatusListener,
+        'zCsQuoHJxVPp895.ZWg9MTIzNDU2LmUudGVzdGluZy5lcHBvLmNsb3Vk',
+      );
+      expect(dispatcher).toBeInstanceOf(DefaultEventDispatcher);
+    });
+  });
 });

src/events/default-event-dispatcher.ts

@@ -4,7 +4,9 @@ import BatchEventProcessor from './batch-event-processor';
 import BatchRetryManager from './batch-retry-manager';
 import EventDelivery from './event-delivery';
 import EventDispatcher, { Event } from './event-dispatcher';
+import NamedEventQueue from './named-event-queue';
 import NetworkStatusListener from './network-status-listener';
+import SdkKeyDecoder from './sdk-key-decoder';
 
 export type EventDispatcherConfig = {
   // target url to deliver events to
@@ -19,6 +21,16 @@ export type EventDispatcherConfig = {
   maxRetries?: number;
 };
 
+// TODO: Have more realistic default batch size based on average event payload size once we have
+//  more concrete data.
+export const DEFAULT_EVENT_DISPATCHER_BATCH_SIZE = 100;
+export const DEFAULT_EVENT_DISPATCHER_CONFIG: Omit<EventDispatcherConfig, 'ingestionUrl'> = {
+  deliveryIntervalMs: 10_000,
+  retryIntervalMs: 5_000,
+  maxRetryDelayMs: 30_000,
+  maxRetries: 3,
+};
+
 /**
  * @internal
  * An {@link EventDispatcher} that, given the provided config settings, delivers events in batches
@@ -37,6 +49,7 @@ export default class DefaultEventDispatcher implements EventDispatcher {
     private readonly networkStatusListener: NetworkStatusListener,
     config: EventDispatcherConfig,
   ) {
+    this.ensureConfigFields(config);
     this.eventDelivery = new EventDelivery(config.ingestionUrl);
     this.retryManager = new BatchRetryManager(this.eventDelivery, {
       retryIntervalMs: config.retryIntervalMs,
@@ -94,4 +107,39 @@ export default class DefaultEventDispatcher implements EventDispatcher {
       this.dispatchTimer = setTimeout(() => this.deliverNextBatch(), this.deliveryIntervalMs);
     }
   }
+
+  private ensureConfigFields(config: EventDispatcherConfig) {
+    if (!config.ingestionUrl) {
+      throw new Error('Missing required ingestionUrl in EventDispatcherConfig');
+    }
+    if (!config.deliveryIntervalMs) {
+      throw new Error('Missing required deliveryIntervalMs in EventDispatcherConfig');
+    }
+    if (!config.retryIntervalMs) {
+      throw new Error('Missing required retryIntervalMs in EventDispatcherConfig');
+    }
+    if (!config.maxRetryDelayMs) {
+      throw new Error('Missing required maxRetryDelayMs in EventDispatcherConfig');
+    }
+  }
+}
+
+/** Creates a new {@link DefaultEventDispatcher} with the provided configuration. */
+export function newDefaultEventDispatcher(
+  eventQueue: NamedEventQueue<unknown>,
+  networkStatusListener: NetworkStatusListener,
+  sdkKey: string,
+  batchSize: number = DEFAULT_EVENT_DISPATCHER_BATCH_SIZE,
+  config: Omit<EventDispatcherConfig, 'ingestionUrl'> = DEFAULT_EVENT_DISPATCHER_CONFIG,
+): EventDispatcher {
+  const sdkKeyDecoder = new SdkKeyDecoder();
+  const ingestionUrl = sdkKeyDecoder.decodeEventIngestionHostName(sdkKey);
+  if (!ingestionUrl) {
+    throw new Error('Unable to parse Event ingestion URL from SDK key');
+  }
+  return new DefaultEventDispatcher(
+    new BatchEventProcessor(eventQueue, batchSize),
+    networkStatusListener,
+    { ...config, ingestionUrl },
+  );
 }

src/events/event-delivery.ts

@@ -1,7 +1,7 @@
 import { logger } from '../application-logger';
 
 export default class EventDelivery {
-  constructor(private ingestionUrl: string) {}
+  constructor(private readonly ingestionUrl: string) {}
 
   async deliver(batch: unknown[]): Promise<boolean> {
     try {

src/index.ts

@@ -29,9 +29,13 @@ import {
 import { HybridConfigurationStore } from './configuration-store/hybrid.store';
 import { MemoryStore, MemoryOnlyConfigurationStore } from './configuration-store/memory.store';
 import * as constants from './constants';
-import ArrayBackedNamedEventQueue from './events/array-backed-named-event-queue';
 import BatchEventProcessor from './events/batch-event-processor';
-import DefaultEventDispatcher from './events/default-event-dispatcher';
+import { BoundedEventQueue } from './events/bounded-event-queue';
+import DefaultEventDispatcher, {
+  DEFAULT_EVENT_DISPATCHER_CONFIG,
+  DEFAULT_EVENT_DISPATCHER_BATCH_SIZE,
+  newDefaultEventDispatcher,
+} from './events/default-event-dispatcher';
 import EventDispatcher from './events/event-dispatcher';
 import NamedEventQueue from './events/named-event-queue';
 import NetworkStatusListener from './events/network-status-listener';
@@ -93,9 +97,13 @@ export {
   BanditSubjectAttributes,
   BanditActions,
 
-  // event queue types
+  // event dispatcher types
   NamedEventQueue,
   EventDispatcher,
+  BoundedEventQueue,
+  DEFAULT_EVENT_DISPATCHER_CONFIG,
+  DEFAULT_EVENT_DISPATCHER_BATCH_SIZE,
+  newDefaultEventDispatcher,
   BatchEventProcessor,
   NetworkStatusListener,
   DefaultEventDispatcher,

yarn.lock

@@ -1831,6 +1831,11 @@ eslint-plugin-promise@^6.0.0:
   resolved "https://registry.npmjs.org/eslint-plugin-promise/-/eslint-plugin-promise-6.0.0.tgz"
   integrity sha512-7GPezalm5Bfi/E22PnQxDWH2iW9GTvAlUNTztemeHb6c1BniSyoeTrM87JkC0wYdi6aQrZX9p2qEiAno8aTcbw==
 
+eslint-plugin-unused-imports@^4.1.4:
+  version "4.1.4"
+  resolved "https://registry.yarnpkg.com/eslint-plugin-unused-imports/-/eslint-plugin-unused-imports-4.1.4.tgz#62ddc7446ccbf9aa7b6f1f0b00a980423cda2738"
+  integrity sha512-YptD6IzQjDardkl0POxnnRBhU1OEePMV0nd6siHaRBbd+lyh6NAhFEobiznKU7kTsSsDeSD62Pe7kAM1b7dAZQ==
+
 [email protected], eslint-scope@^5.1.1:
   version "5.1.1"
   resolved "https://registry.npmjs.org/eslint-scope/-/eslint-scope-5.1.1.tgz"