[FSSDK-9605] add support for configurable closing event dispatcher (#874)

Author: raju-opti

packages/event-processor/__tests__/v1EventProcessor.spec.ts

@@ -525,5 +525,41 @@ describe('LogTierV1EventProcessor', () => {
         params: makeBatchedEventV1(impressionEvents),
       })
     })
+
+    it('should use the provided closingDispatcher to dispatch events on stop', async () => {
+      const dispatcher = {
+        dispatchEvent: jest.fn(),
+      }
+
+      const closingDispatcher = {
+        dispatchEvent: jest.fn(),
+      }
+
+      const processor = new LogTierV1EventProcessor({
+        dispatcher,
+        closingDispatcher,
+        flushInterval: 100000,
+        batchSize: 20,
+      });
+
+      processor.start()
+
+      const events: ReturnType<typeof createImpressionEvent>[] = [];
+
+      for (let i = 0; i < 4; i++) {
+        const event = createImpressionEvent();
+        processor.process(event);
+        events.push(event);
+      }
+
+      processor.stop();
+      jest.runAllTimers();
+
+      expect(dispatcher.dispatchEvent).not.toHaveBeenCalled();
+      expect(closingDispatcher.dispatchEvent).toHaveBeenCalledTimes(1);
+
+      const [data] = closingDispatcher.dispatchEvent.mock.calls[0];
+      expect(data.params).toEqual(makeBatchedEventV1(events));
+    })
   })
 })

packages/event-processor/src/eventProcessor.ts

@@ -1,5 +1,5 @@
 /**
- * Copyright 2019-2020, Optimizely
+ * Copyright 2019-2020, 2023 Optimizely
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,7 +17,7 @@
 import { Managed } from './managed'
 import { ConversionEvent, ImpressionEvent } from './events'
 import { EventV1Request } from './eventDispatcher'
-import { EventQueue, DefaultEventQueue, SingleEventQueue } from './eventQueue'
+import { EventQueue, DefaultEventQueue, SingleEventQueue, EventQueueSink } from './eventQueue'
 import { getLogger } from '@optimizely/js-sdk-logging'
 import { NOTIFICATION_TYPES, NotificationCenter } from '@optimizely/js-sdk-utils'
 
@@ -56,14 +56,21 @@ export function validateAndGetBatchSize(batchSize: number): number {
   return batchSize
 }
 
-export function getQueue(batchSize: number, flushInterval: number, sink: any, batchComparator: any): EventQueue<ProcessableEvent> {
+export function getQueue(
+  batchSize: number, 
+  flushInterval: number, 
+  batchComparator: any, 
+  sink: EventQueueSink<ProcessableEvent>, 
+  closingSink?: EventQueueSink<ProcessableEvent>,
+): EventQueue<ProcessableEvent> {
   let queue: EventQueue<ProcessableEvent>
   if (batchSize > 1) {
     queue = new DefaultEventQueue<ProcessableEvent>({
       flushInterval,
       maxQueueSize: batchSize,
-      sink,
       batchComparator,
+      sink,
+      closingSink,
     })
   } else {
     queue = new SingleEventQueue({ sink })

packages/event-processor/src/eventQueue.ts

@@ -1,5 +1,5 @@
 /**
- * Copyright 2019, Optimizely
+ * Copyright 2019, 2023 Optimizely
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -87,6 +87,7 @@ export class DefaultEventQueue<K> implements EventQueue<K> {
   private buffer: K[]
   private maxQueueSize: number
   private sink: EventQueueSink<K>
+  private closingSink?: EventQueueSink<K>
   // batchComparator is called to determine whether two events can be included
   // together in the same batch
   private batchComparator: (eventA: K, eventB: K) => boolean
@@ -96,16 +97,19 @@ export class DefaultEventQueue<K> implements EventQueue<K> {
     flushInterval,
     maxQueueSize,
     sink,
+    closingSink,
     batchComparator,
   }: {
     flushInterval: number
     maxQueueSize: number
     sink: EventQueueSink<K>
+    closingSink?: EventQueueSink<K>
     batchComparator: (eventA: K, eventB: K) => boolean
   }) {
     this.buffer = []
     this.maxQueueSize = Math.max(maxQueueSize, 1)
     this.sink = sink
+    this.closingSink = closingSink;
     this.batchComparator = batchComparator
     this.timer = new Timer({
       callback: this.flush.bind(this),
@@ -121,7 +125,7 @@ export class DefaultEventQueue<K> implements EventQueue<K> {
 
   stop(): Promise<any> {
     this.started = false
-    const result = this.sink(this.buffer)
+    const result = this.closingSink ? this.closingSink(this.buffer) : this.sink(this.buffer);
     this.buffer = []
     this.timer.stop()
     return result

packages/event-processor/src/v1/v1EventProcessor.react_native.ts

@@ -1,5 +1,5 @@
 /**
- * Copyright 2020, Optimizely
+ * Copyright 2020, 2023, Optimizely
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -102,7 +102,7 @@ export class LogTierV1EventProcessor implements EventProcessor {
 
     flushInterval = validateAndGetFlushInterval(flushInterval)
     batchSize = validateAndGetBatchSize(batchSize)
-    this.queue = getQueue(batchSize, flushInterval, this.drainQueue.bind(this), areEventContextsEqual)
+    this.queue = getQueue(batchSize, flushInterval, areEventContextsEqual, this.drainQueue.bind(this))
     this.pendingEventsStore = new ReactNativeEventsStore(maxQueueSize, PENDING_EVENTS_STORE_KEY)
     this.eventBufferStore = new ReactNativeEventsStore(maxQueueSize, EVENT_BUFFER_STORE_KEY)
   }

packages/event-processor/src/v1/v1EventProcessor.ts

@@ -1,5 +1,5 @@
 /**
- * Copyright 2019-2020, Optimizely
+ * Copyright 2019-2020, 2023 Optimizely
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -36,31 +36,41 @@ const logger = getLogger('LogTierV1EventProcessor')
 
 export class LogTierV1EventProcessor implements EventProcessor {
   private dispatcher: EventDispatcher
+  private closingDispatcher?: EventDispatcher
   private queue: EventQueue<ProcessableEvent>
   private notificationCenter?: NotificationCenter
   private requestTracker: RequestTracker
 
   constructor({
     dispatcher,
+    closingDispatcher,
     flushInterval = DEFAULT_FLUSH_INTERVAL,
     batchSize = DEFAULT_BATCH_SIZE,
     notificationCenter,
   }: {
     dispatcher: EventDispatcher
+    closingDispatcher?: EventDispatcher
     flushInterval?: number
     batchSize?: number
     notificationCenter?: NotificationCenter
   }) {
     this.dispatcher = dispatcher
+    this.closingDispatcher = closingDispatcher
     this.notificationCenter = notificationCenter
     this.requestTracker = new RequestTracker()
 
     flushInterval = validateAndGetFlushInterval(flushInterval)
     batchSize = validateAndGetBatchSize(batchSize)
-    this.queue = getQueue(batchSize, flushInterval, this.drainQueue.bind(this), areEventContextsEqual)
+    this.queue = getQueue(
+      batchSize,
+      flushInterval, 
+      areEventContextsEqual,
+      this.drainQueue.bind(this, false),
+      this.drainQueue.bind(this, true),
+    )
   }
 
-  drainQueue(buffer: ProcessableEvent[]): Promise<void> {
+  private drainQueue(useClosingDispatcher: boolean, buffer: ProcessableEvent[]): Promise<void> {
     const reqPromise = new Promise<void>(resolve => {
       logger.debug('draining queue with %s events', buffer.length)
 
@@ -70,7 +80,10 @@ export class LogTierV1EventProcessor implements EventProcessor {
       }
 
       const formattedEvent = formatEvents(buffer)
-      this.dispatcher.dispatchEvent(formattedEvent, () => {
+      const dispatcher = useClosingDispatcher && this.closingDispatcher
+        ? this.closingDispatcher : this.dispatcher;
+
+      dispatcher.dispatchEvent(formattedEvent, () => {
         resolve()
       })
       sendEventNotification(this.notificationCenter, formattedEvent)