[FSSDK-10941] event processor files and directories cleanup - part 2 (#967)

Author: raju-opti

lib/event_processor/batch_event_processor.react_native.spec.ts

@@ -50,7 +50,7 @@ import { getMockRepeater } from '../tests/mock/mock_repeater';
 import { getMockAsyncCache } from '../tests/mock/mock_cache';
 
 import { EventWithId } from './batch_event_processor';
-import { formatEvents } from './event_builder/build_event_v1';
+import { buildLogEvent } from './event_builder/build_event_v1';
 import { createImpressionEvent } from '../tests/mock/create_event';
 import { ProcessableEvent } from './event_processor';
 
@@ -138,7 +138,7 @@ describe('ReactNativeNetInfoEventProcessor', () => {
 
     await exhaustMicrotasks();
 
-    expect(eventDispatcher.dispatchEvent).toHaveBeenCalledWith(formatEvents(events));
+    expect(eventDispatcher.dispatchEvent).toHaveBeenCalledWith(buildLogEvent(events));
   });
 
   it('should unsubscribe from netinfo listener when stopped', async () => {

lib/event_processor/batch_event_processor.spec.ts

@@ -19,7 +19,7 @@ import { EventWithId, BatchEventProcessor } from './batch_event_processor';
 import { getMockSyncCache } from '../tests/mock/mock_cache';
 import { createImpressionEvent } from '../tests/mock/create_event';
 import { ProcessableEvent } from './event_processor';
-import { formatEvents } from './event_builder/build_event_v1';
+import { buildLogEvent } from './event_builder/build_event_v1';
 import { resolvablePromise } from '../utils/promise/resolvablePromise';
 import { advanceTimersByTime } from  '../../tests/testUtils';
 import { getMockLogger } from '../tests/mock/mock_logger';
@@ -139,9 +139,9 @@ describe('QueueingEventProcessor', async () => {
       await exhaustMicrotasks();
 
       expect(mockDispatch).toHaveBeenCalledTimes(3);
-      expect(mockDispatch.mock.calls[0][0]).toEqual(formatEvents([events[0], events[1]]));
-      expect(mockDispatch.mock.calls[1][0]).toEqual(formatEvents([events[2], events[3]]));
-      expect(mockDispatch.mock.calls[2][0]).toEqual(formatEvents([events[4]]));
+      expect(mockDispatch.mock.calls[0][0]).toEqual(buildLogEvent([events[0], events[1]]));
+      expect(mockDispatch.mock.calls[1][0]).toEqual(buildLogEvent([events[2], events[3]]));
+      expect(mockDispatch.mock.calls[2][0]).toEqual(buildLogEvent([events[4]]));
     });
   });
 
@@ -202,7 +202,7 @@ describe('QueueingEventProcessor', async () => {
       await processor.process(event);
 
       expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(1);
-      expect(eventDispatcher.dispatchEvent.mock.calls[0][0]).toEqual(formatEvents(events));
+      expect(eventDispatcher.dispatchEvent.mock.calls[0][0]).toEqual(buildLogEvent(events));
 
       events = [event];
       for(let i = 101; i < 200; i++) {
@@ -217,7 +217,7 @@ describe('QueueingEventProcessor', async () => {
       await processor.process(event);
 
       expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(2);
-      expect(eventDispatcher.dispatchEvent.mock.calls[1][0]).toEqual(formatEvents(events));
+      expect(eventDispatcher.dispatchEvent.mock.calls[1][0]).toEqual(buildLogEvent(events));
     });
 
     it('should flush queue is context of the new event is different and enqueue the new event', async () => {
@@ -250,11 +250,11 @@ describe('QueueingEventProcessor', async () => {
       await processor.process(newEvent);
 
       expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(1);
-      expect(eventDispatcher.dispatchEvent.mock.calls[0][0]).toEqual(formatEvents(events));
+      expect(eventDispatcher.dispatchEvent.mock.calls[0][0]).toEqual(buildLogEvent(events));
 
       await dispatchRepeater.execute(0);
       expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(2);
-      expect(eventDispatcher.dispatchEvent.mock.calls[1][0]).toEqual(formatEvents([newEvent]));
+      expect(eventDispatcher.dispatchEvent.mock.calls[1][0]).toEqual(buildLogEvent([newEvent]));
     });
 
     it('should store the event in the eventStore with increasing ids', async () => {
@@ -313,7 +313,7 @@ describe('QueueingEventProcessor', async () => {
     await dispatchRepeater.execute(0);
 
     expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(1);
-    expect(eventDispatcher.dispatchEvent.mock.calls[0][0]).toEqual(formatEvents(events));
+    expect(eventDispatcher.dispatchEvent.mock.calls[0][0]).toEqual(buildLogEvent(events));
 
     events = [];
     for(let i = 1; i < 15; i++) {
@@ -324,7 +324,7 @@ describe('QueueingEventProcessor', async () => {
 
     await dispatchRepeater.execute(0);
     expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(2);
-    expect(eventDispatcher.dispatchEvent.mock.calls[1][0]).toEqual(formatEvents(events));
+    expect(eventDispatcher.dispatchEvent.mock.calls[1][0]).toEqual(buildLogEvent(events));
   });
 
   it('should not retry failed dispatch if retryConfig is not provided', async () => {
@@ -397,7 +397,7 @@ describe('QueueingEventProcessor', async () => {
     expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(4);
     expect(backoffController.backoff).toHaveBeenCalledTimes(3);
 
-    const request = formatEvents(events);
+    const request = buildLogEvent(events);
     for(let i = 0; i < 4; i++) {
       expect(eventDispatcher.dispatchEvent.mock.calls[i][0]).toEqual(request);
     }
@@ -444,7 +444,7 @@ describe('QueueingEventProcessor', async () => {
     expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(201);
     expect(backoffController.backoff).toHaveBeenCalledTimes(200);
 
-    const request = formatEvents(events);
+    const request = buildLogEvent(events);
     for(let i = 0; i < 201; i++) {
       expect(eventDispatcher.dispatchEvent.mock.calls[i][0]).toEqual(request);
     }
@@ -723,7 +723,7 @@ describe('QueueingEventProcessor', async () => {
       await exhaustMicrotasks();
 
       expect(mockDispatch).toHaveBeenCalledTimes(1);
-      expect(mockDispatch.mock.calls[0][0]).toEqual(formatEvents(failedEvents));
+      expect(mockDispatch.mock.calls[0][0]).toEqual(buildLogEvent(failedEvents));
 
       const eventsInStore = [...cache.getAll().values()].sort((a, b) => a.id < b.id ? -1 : 1).map(e => e.event);
       expect(eventsInStore).toEqual(expect.arrayContaining([
@@ -761,7 +761,7 @@ describe('QueueingEventProcessor', async () => {
       dispatchRepeater.execute(0);
       await exhaustMicrotasks();
       expect(mockDispatch).toHaveBeenCalledTimes(1);
-      expect(mockDispatch.mock.calls[0][0]).toEqual(formatEvents([eventA, eventB]));
+      expect(mockDispatch.mock.calls[0][0]).toEqual(buildLogEvent([eventA, eventB]));
 
       const failedEvents: ProcessableEvent[] = [];
 
@@ -776,7 +776,7 @@ describe('QueueingEventProcessor', async () => {
       await exhaustMicrotasks();
 
       expect(mockDispatch).toHaveBeenCalledTimes(2);
-      expect(mockDispatch.mock.calls[1][0]).toEqual(formatEvents(failedEvents));
+      expect(mockDispatch.mock.calls[1][0]).toEqual(buildLogEvent(failedEvents));
 
       mockResult2.resolve({});
       await exhaustMicrotasks();
@@ -826,10 +826,10 @@ describe('QueueingEventProcessor', async () => {
       // events  0 1 4 5 6 7 have one context, and 2 3 have different context
       // batches should be [0, 1], [2, 3], [4, 5, 6], [7]
       expect(mockDispatch).toHaveBeenCalledTimes(4);
-      expect(mockDispatch.mock.calls[0][0]).toEqual(formatEvents([failedEvents[0], failedEvents[1]]));
-      expect(mockDispatch.mock.calls[1][0]).toEqual(formatEvents([failedEvents[2], failedEvents[3]]));
-      expect(mockDispatch.mock.calls[2][0]).toEqual(formatEvents([failedEvents[4], failedEvents[5], failedEvents[6]]));
-      expect(mockDispatch.mock.calls[3][0]).toEqual(formatEvents([failedEvents[7]]));
+      expect(mockDispatch.mock.calls[0][0]).toEqual(buildLogEvent([failedEvents[0], failedEvents[1]]));
+      expect(mockDispatch.mock.calls[1][0]).toEqual(buildLogEvent([failedEvents[2], failedEvents[3]]));
+      expect(mockDispatch.mock.calls[2][0]).toEqual(buildLogEvent([failedEvents[4], failedEvents[5], failedEvents[6]]));
+      expect(mockDispatch.mock.calls[3][0]).toEqual(buildLogEvent([failedEvents[7]]));
     });
   });
 
@@ -873,7 +873,7 @@ describe('QueueingEventProcessor', async () => {
       await exhaustMicrotasks();
 
       expect(mockDispatch).toHaveBeenCalledTimes(1);
-      expect(mockDispatch.mock.calls[0][0]).toEqual(formatEvents(failedEvents));
+      expect(mockDispatch.mock.calls[0][0]).toEqual(buildLogEvent(failedEvents));
 
       const eventsInStore = [...cache.getAll().values()].sort((a, b) => a.id < b.id ? -1 : 1).map(e => e.event);
       expect(eventsInStore).toEqual(expect.arrayContaining([
@@ -913,7 +913,7 @@ describe('QueueingEventProcessor', async () => {
       dispatchRepeater.execute(0);
       await exhaustMicrotasks();
       expect(mockDispatch).toHaveBeenCalledTimes(1);
-      expect(mockDispatch.mock.calls[0][0]).toEqual(formatEvents([eventA, eventB]));
+      expect(mockDispatch.mock.calls[0][0]).toEqual(buildLogEvent([eventA, eventB]));
 
       const failedEvents: ProcessableEvent[] = [];
 
@@ -928,7 +928,7 @@ describe('QueueingEventProcessor', async () => {
       await exhaustMicrotasks();
 
       expect(mockDispatch).toHaveBeenCalledTimes(2);
-      expect(mockDispatch.mock.calls[1][0]).toEqual(formatEvents(failedEvents));
+      expect(mockDispatch.mock.calls[1][0]).toEqual(buildLogEvent(failedEvents));
 
       mockResult2.resolve({});
       await exhaustMicrotasks();
@@ -980,10 +980,10 @@ describe('QueueingEventProcessor', async () => {
       // events  0 1 4 5 6 7 have one context, and 2 3 have different context
       // batches should be [0, 1], [2, 3], [4, 5, 6], [7]
       expect(mockDispatch).toHaveBeenCalledTimes(4);
-      expect(mockDispatch.mock.calls[0][0]).toEqual(formatEvents([failedEvents[0], failedEvents[1]]));
-      expect(mockDispatch.mock.calls[1][0]).toEqual(formatEvents([failedEvents[2], failedEvents[3]]));
-      expect(mockDispatch.mock.calls[2][0]).toEqual(formatEvents([failedEvents[4], failedEvents[5], failedEvents[6]]));
-      expect(mockDispatch.mock.calls[3][0]).toEqual(formatEvents([failedEvents[7]]));
+      expect(mockDispatch.mock.calls[0][0]).toEqual(buildLogEvent([failedEvents[0], failedEvents[1]]));
+      expect(mockDispatch.mock.calls[1][0]).toEqual(buildLogEvent([failedEvents[2], failedEvents[3]]));
+      expect(mockDispatch.mock.calls[2][0]).toEqual(buildLogEvent([failedEvents[4], failedEvents[5], failedEvents[6]]));
+      expect(mockDispatch.mock.calls[3][0]).toEqual(buildLogEvent([failedEvents[7]]));
     });
   });
 
@@ -1012,7 +1012,7 @@ describe('QueueingEventProcessor', async () => {
     await dispatchRepeater.execute(0);
 
     expect(dispatchListener).toHaveBeenCalledTimes(1);
-    expect(dispatchListener.mock.calls[0][0]).toEqual(formatEvents([event, event2]));
+    expect(dispatchListener.mock.calls[0][0]).toEqual(buildLogEvent([event, event2]));
   });
 
   it('should remove event handler when function returned from onDispatch is called', async () => {
@@ -1041,7 +1041,7 @@ describe('QueueingEventProcessor', async () => {
     await dispatchRepeater.execute(0);
 
     expect(dispatchListener).toHaveBeenCalledTimes(1);
-    expect(dispatchListener.mock.calls[0][0]).toEqual(formatEvents([event, event2]));
+    expect(dispatchListener.mock.calls[0][0]).toEqual(buildLogEvent([event, event2]));
 
     unsub();
 
@@ -1119,7 +1119,7 @@ describe('QueueingEventProcessor', async () => {
 
       processor.stop();
       expect(closingEventDispatcher.dispatchEvent).toHaveBeenCalledTimes(1);
-      expect(closingEventDispatcher.dispatchEvent).toHaveBeenCalledWith(formatEvents(events));
+      expect(closingEventDispatcher.dispatchEvent).toHaveBeenCalledWith(buildLogEvent(events));
     });
 
     it('should cancel retry of active dispatches', async () => {

lib/event_processor/batch_event_processor.ts

@@ -16,8 +16,8 @@
 
 import { EventProcessor, ProcessableEvent } from "./event_processor";
 import { Cache } from "../utils/cache/cache";
-import { EventDispatcher, EventDispatcherResponse, EventV1Request } from "./event_dispatcher";
-import { formatEvents } from "./event_builder/build_event_v1";
+import { EventDispatcher, EventDispatcherResponse, LogEvent } from "./event_dispatcher";
+import { buildLogEvent } from "./event_builder/log_event";
 import { BackoffController, ExponentialBackoff, IntervalRepeater, Repeater } from "../utils/repeater/repeater";
 import { LoggerFacade } from "../modules/logging";
 import { BaseService, ServiceState, StartupLog } from "../service";
@@ -26,7 +26,7 @@ import { RunResult, runWithRetry } from "../utils/executor/backoff_retry_runner"
 import { isSuccessStatusCode } from "../utils/http_request_handler/http_util";
 import { EventEmitter } from "../utils/event_emitter/event_emitter";
 import { IdGenerator } from "../utils/id_generator";
-import { areEventContextsEqual } from "./events";
+import { areEventContextsEqual } from "./event_builder/user_event";
 
 export type EventWithId = {
   id: string;
@@ -51,7 +51,7 @@ export type BatchEventProcessorConfig = {
 };
 
 type EventBatch = {
-  request: EventV1Request,
+  request: LogEvent,
   ids: string[],
 }
 
@@ -66,7 +66,7 @@ export class BatchEventProcessor extends BaseService implements EventProcessor {
   private idGenerator: IdGenerator = new IdGenerator();
   private runningTask: Map<string, RunResult<EventDispatcherResponse>> = new Map();
   private dispatchingEventIds: Set<string> = new Set();
-  private eventEmitter: EventEmitter<{ dispatch: EventV1Request }> = new EventEmitter();
+  private eventEmitter: EventEmitter<{ dispatch: LogEvent }> = new EventEmitter();
   private retryConfig?: RetryConfig;
 
   constructor(config: BatchEventProcessorConfig) {
@@ -85,7 +85,7 @@ export class BatchEventProcessor extends BaseService implements EventProcessor {
     this.failedEventRepeater?.setTask(() => this.retryFailedEvents());
   }
 
-  onDispatch(handler: Consumer<EventV1Request>): Fn {
+  onDispatch(handler: Consumer<LogEvent>): Fn {
     return this.eventEmitter.on('dispatch', handler);
   }
 
@@ -119,7 +119,7 @@ export class BatchEventProcessor extends BaseService implements EventProcessor {
       if (currentBatch.length === this.batchSize ||
            (currentBatch.length > 0 && !areEventContextsEqual(currentBatch[0].event, event.event))) {
         batches.push({
-          request: formatEvents(currentBatch.map((e) => e.event)),
+          request: buildLogEvent(currentBatch.map((e) => e.event)),
           ids: currentBatch.map((e) => e.id),
         });
         currentBatch = [];
@@ -129,7 +129,7 @@ export class BatchEventProcessor extends BaseService implements EventProcessor {
 
     if (currentBatch.length > 0) {
       batches.push({
-        request: formatEvents(currentBatch.map((e) => e.event)),
+        request: buildLogEvent(currentBatch.map((e) => e.event)),
         ids: currentBatch.map((e) => e.id),
       });
     }
@@ -153,10 +153,10 @@ export class BatchEventProcessor extends BaseService implements EventProcessor {
     });
 
     this.eventQueue = [];
-    return { request: formatEvents(events), ids };
+    return { request: buildLogEvent(events), ids };
   }
 
-  private async executeDispatch(request: EventV1Request, closing = false): Promise<EventDispatcherResponse> {
+  private async executeDispatch(request: LogEvent, closing = false): Promise<EventDispatcherResponse> {
     const dispatcher = closing && this.closingEventDispatcher ? this.closingEventDispatcher : this.eventDispatcher;
     return dispatcher.dispatchEvent(request).then((res) => {
       if (res.statusCode && !isSuccessStatusCode(res.statusCode)) {

lib/event_processor/default_dispatcher.spec.ts

@@ -15,9 +15,9 @@
  */
 import { expect, vi, describe, it } from 'vitest';
 import { DefaultEventDispatcher } from './default_dispatcher';
-import { EventV1 } from './event_builder/build_event_v1';
+import { EventBatch } from './event_builder/build_event_v1';
 
-const getEvent = (): EventV1 => {
+const getEvent = (): EventBatch => {
   return {
     account_id: 'string',
     project_id: 'string',

lib/event_processor/default_dispatcher.ts

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 import { RequestHandler } from '../utils/http_request_handler/http';
-import { EventDispatcher, EventDispatcherResponse, EventV1Request } from './event_dispatcher';
+import { EventDispatcher, EventDispatcherResponse, LogEvent } from './event_dispatcher';
 
 export class DefaultEventDispatcher implements EventDispatcher {
   private requestHandler: RequestHandler;
@@ -24,7 +24,7 @@ export class DefaultEventDispatcher implements EventDispatcher {
   }
 
   async dispatchEvent(
-    eventObj: EventV1Request
+    eventObj: LogEvent
   ): Promise<EventDispatcherResponse> {
     // Non-POST requests not supported
     if (eventObj.httpVerb !== 'POST') {