[FSSDK-9605] add sendBeacon event dispatcher support (#866)

Author: raju-opti

lib/index.browser.ts

@@ -19,6 +19,7 @@ import { LocalStoragePendingEventsDispatcher } from './modules/event_processor';
 import configValidator from './utils/config_validator';
 import defaultErrorHandler from './plugins/error_handler';
 import defaultEventDispatcher from './plugins/event_dispatcher/index.browser';
+import sendBeaconEventDispatcher from './plugins/event_dispatcher/send_beacon_dispatcher';
 import * as enums from './utils/enums';
 import * as loggerPlugin from './plugins/logger';
 import eventProcessorConfigValidator from './utils/event_processor_config_validator';
@@ -90,6 +91,12 @@ const createInstance = function(config: Config): Client | null {
       eventDispatcher = config.eventDispatcher;
     }
 
+    let closingDispatcher = config.closingEventDispatcher;
+
+    if (!config.eventDispatcher && !closingDispatcher && window.navigator && 'sendBeacon' in window.navigator) {
+      closingDispatcher = sendBeaconEventDispatcher;
+    }
+
     let eventBatchSize = config.eventBatchSize;
     let eventFlushInterval = config.eventFlushInterval;
 
@@ -111,6 +118,7 @@ const createInstance = function(config: Config): Client | null {
 
     const eventProcessorConfig = {
       dispatcher: eventDispatcher,
+      closingDispatcher,
       flushInterval: eventFlushInterval,
       batchSize: eventBatchSize,
       maxQueueSize: config.eventMaxQueueSize || DEFAULT_EVENT_MAX_QUEUE_SIZE,
@@ -172,6 +180,7 @@ export {
   loggerPlugin as logging,
   defaultErrorHandler as errorHandler,
   defaultEventDispatcher as eventDispatcher,
+  sendBeaconEventDispatcher,
   enums,
   setLogHandler as setLogger,
   setLogLevel,
@@ -189,6 +198,7 @@ export default {
   logging: loggerPlugin,
   errorHandler: defaultErrorHandler,
   eventDispatcher: defaultEventDispatcher,
+  sendBeaconEventDispatcher,
   enums,
   setLogger: setLogHandler,
   setLogLevel,

lib/modules/event_processor/eventProcessor.ts

@@ -1,5 +1,5 @@
 /**
- * Copyright 2022, Optimizely
+ * Copyright 2022-2023 Optimizely
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -57,13 +57,20 @@ export function validateAndGetBatchSize(batchSize: number): number {
   return batchSize
 }
 
-export function getQueue(batchSize: number, flushInterval: number, sink: EventQueueSink<ProcessableEvent>, batchComparator: (eventA: ProcessableEvent, eventB: ProcessableEvent) => boolean ): EventQueue<ProcessableEvent> {
+export function getQueue(
+  batchSize: number, 
+  flushInterval: number, 
+  batchComparator: (eventA: ProcessableEvent, eventB: ProcessableEvent) => boolean,
+  sink: EventQueueSink<ProcessableEvent>,
+  closingSink?: EventQueueSink<ProcessableEvent>
+): EventQueue<ProcessableEvent> {
   let queue: EventQueue<ProcessableEvent>
   if (batchSize > 1) {
     queue = new DefaultEventQueue<ProcessableEvent>({
       flushInterval,
       maxQueueSize: batchSize,
       sink,
+      closingSink,
       batchComparator,
     })
   } else {

lib/modules/event_processor/eventQueue.ts

@@ -1,5 +1,5 @@
 /**
- * Copyright 2022, Optimizely
+ * Copyright 2022-2023, Optimizely
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -14,149 +14,149 @@
  * limitations under the License.
  */
 
-import { getLogger } from '../logging'
+import { getLogger } from '../logging';
 // TODO change this to use Managed from js-sdk-models when available
-import { Managed } from './managed'
+import { Managed } from './managed';
 
-const logger = getLogger('EventProcessor')
+const logger = getLogger('EventProcessor');
 
-export type EventQueueSink<K> = (buffer: K[]) => Promise<any>
+export type EventQueueSink<K> = (buffer: K[]) => Promise<any>;
 
 export interface EventQueue<K> extends Managed {
-  enqueue(event: K): void
+  enqueue(event: K): void;
 }
 
 export interface EventQueueFactory<K> {
-  createEventQueue(config: {
-    sink: EventQueueSink<K>
-    flushInterval: number
-    maxQueueSize: number
-  }): EventQueue<K>
+  createEventQueue(config: { sink: EventQueueSink<K>, flushInterval: number, maxQueueSize: number }): EventQueue<K>;
 }
 
 class Timer {
-  private timeout: number
-  private callback: () => void
-  private timeoutId?: number
+  private timeout: number;
+  private callback: () => void;
+  private timeoutId?: number;
 
   constructor({ timeout, callback }: { timeout: number; callback: () => void }) {
-    this.timeout = Math.max(timeout, 0)
-    this.callback = callback
+    this.timeout = Math.max(timeout, 0);
+    this.callback = callback;
   }
 
   start(): void {
-    this.timeoutId = setTimeout(this.callback, this.timeout) as any
+    this.timeoutId = setTimeout(this.callback, this.timeout) as any;
   }
 
   refresh(): void {
-    this.stop()
-    this.start()
+    this.stop();
+    this.start();
   }
 
   stop(): void {
     if (this.timeoutId) {
-      clearTimeout(this.timeoutId as any)
+      clearTimeout(this.timeoutId as any);
     }
   }
 }
 
 export class SingleEventQueue<K> implements EventQueue<K> {
-  private sink: EventQueueSink<K>
+  private sink: EventQueueSink<K>;
 
   constructor({ sink }: { sink: EventQueueSink<K> }) {
-    this.sink = sink
+    this.sink = sink;
   }
 
   start(): Promise<any> {
     // no-op
-    return Promise.resolve()
+    return Promise.resolve();
   }
 
   stop(): Promise<any> {
     // no-op
-    return Promise.resolve()
+    return Promise.resolve();
   }
 
   enqueue(event: K): void {
-    this.sink([event])
+    this.sink([event]);
   }
 }
 
 export class DefaultEventQueue<K> implements EventQueue<K> {
   // expose for testing
-  public timer: Timer
-  private buffer: K[]
-  private maxQueueSize: number
-  private sink: EventQueueSink<K>
+  public timer: Timer;
+  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
-  private started: boolean
+  private batchComparator: (eventA: K, eventB: K) => boolean;
+  private started: boolean;
 
   constructor({
     flushInterval,
     maxQueueSize,
     sink,
+    closingSink,
     batchComparator,
   }: {
-    flushInterval: number
-    maxQueueSize: number
-    sink: EventQueueSink<K>
-    batchComparator: (eventA: K, eventB: K) => boolean
+    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.batchComparator = batchComparator
+    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),
       timeout: flushInterval,
-    })
-    this.started = false
+    });
+    this.started = false;
   }
 
   start(): Promise<any> {
-    this.started = true
+    this.started = true;
     // dont start the timer until the first event is enqueued
 
     return Promise.resolve();
   }
 
   stop(): Promise<any> {
-    this.started = false
-    const result = this.sink(this.buffer)
-    this.buffer = []
-    this.timer.stop()
-    return result
+    this.started = false;
+    const result = this.closingSink ? this.closingSink(this.buffer) : this.sink(this.buffer);
+    this.buffer = [];
+    this.timer.stop();
+    return result;
   }
 
   enqueue(event: K): void {
     if (!this.started) {
-      logger.warn('Queue is stopped, not accepting event')
-      return
+      logger.warn('Queue is stopped, not accepting event');
+      return;
     }
 
     // If new event cannot be included into the current batch, flush so it can
     // be in its own new batch.
-    const bufferedEvent: K | undefined = this.buffer[0]
+    const bufferedEvent: K | undefined = this.buffer[0];
     if (bufferedEvent && !this.batchComparator(bufferedEvent, event)) {
-      this.flush()
+      this.flush();
     }
 
     // start the timer when the first event is put in
     if (this.buffer.length === 0) {
-      this.timer.refresh()
+      this.timer.refresh();
     }
-    this.buffer.push(event)
+    this.buffer.push(event);
 
     if (this.buffer.length >= this.maxQueueSize) {
-      this.flush()
+      this.flush();
     }
   }
 
-  flush() : void {
-    this.sink(this.buffer)
-    this.buffer = []
-    this.timer.stop()
+  flush(): void {
+    this.sink(this.buffer);
+    this.buffer = [];
+    this.timer.stop();
   }
 }

lib/modules/event_processor/v1/v1EventProcessor.react_native.ts

@@ -1,5 +1,5 @@
 /**
- * Copyright 2022, Optimizely
+ * Copyright 2022-2023, Optimizely
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -104,7 +104,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)
   }

lib/modules/event_processor/v1/v1EventProcessor.ts

@@ -1,5 +1,5 @@
 /**
- * Copyright 2022, Optimizely
+ * Copyright 2022-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?: NotificationSender
   private requestTracker: RequestTracker
 
   constructor({
     dispatcher,
+    closingDispatcher,
     flushInterval = DEFAULT_FLUSH_INTERVAL,
     batchSize = DEFAULT_BATCH_SIZE,
     notificationCenter,
   }: {
     dispatcher: EventDispatcher
+    closingDispatcher?: EventDispatcher
     flushInterval?: number
     batchSize?: number
     notificationCenter?: NotificationSender
   }) {
     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)