[FSSDK-10642] Refactor batch event processor

lib/core/event_builder/build_event_v1.ts

@@ -17,7 +17,7 @@ import {
   EventTags,
   ConversionEvent,
   ImpressionEvent,
-} from '../../event_processor';
+} from '../../event_processor/events';
 
 import { Event } from '../../shared_types';
 

lib/core/event_builder/index.ts

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 import { LoggerFacade } from '../../modules/logging';
-import { EventV1 as CommonEventParams } from '../../event_processor';
+import { EventV1 as CommonEventParams } from '../../event_processor/v1/buildEventV1';
 
 import fns from '../../utils/fns';
 import { CONTROL_ATTRIBUTES, RESERVED_EVENT_KEYWORDS } from '../../utils/enums';

lib/event_processor/batch_event_processor.react_native.spec.ts

@@ -0,0 +1,155 @@
+import { vi, describe, it, expect, beforeEach } from 'vitest';
+
+const mockNetInfo = vi.hoisted(() => {
+  const netInfo = {
+    listeners: [],
+    unsubs: [],
+    addEventListener(fn: any) {
+      this.listeners.push(fn);
+      const unsub = vi.fn();
+      this.unsubs.push(unsub);
+      return unsub;
+    },
+    pushState(state: boolean) {
+      for (const listener of this.listeners) {
+        listener({ isInternetReachable: state });
+      }
+    },
+    clear() {
+      this.listeners = [];
+      this.unsubs = [];
+    }
+  };
+  return netInfo;
+});
+
+vi.mock('@react-native-community/netinfo', () => {
+  return {
+    addEventListener: mockNetInfo.addEventListener.bind(mockNetInfo),
+  };
+});
+
+import { ReactNativeNetInfoEventProcessor } from './batch_event_processor.react_native';
+import { getMockLogger } from '../tests/mock/mock_logger';
+import { getMockRepeater } from '../tests/mock/mock_repeater';
+import { getMockAsyncCache } from '../tests/mock/mock_cache';
+
+import { EventWithId } from './batch_event_processor';
+import { EventDispatcher } from './eventDispatcher';
+import { formatEvents } from './v1/buildEventV1';
+import { createImpressionEvent } from '../tests/mock/create_event';
+import { ProcessableEvent } from './eventProcessor';
+
+const getMockDispatcher = () => {
+  return {
+    dispatchEvent: vi.fn(),
+  };
+};
+
+const exhaustMicrotasks = async (loop = 100) => {
+  for(let i = 0; i < loop; i++) {
+    await Promise.resolve();
+  }
+}
+
+
+describe('ReactNativeNetInfoEventProcessor', () => {
+  beforeEach(() => {
+    mockNetInfo.clear();
+  });
+
+  it('should not retry failed events when reachable state does not change', async () => {
+    const eventDispatcher = getMockDispatcher();
+    const dispatchRepeater = getMockRepeater();
+    const failedEventRepeater = getMockRepeater();
+
+    const cache = getMockAsyncCache<EventWithId>();
+    const events: ProcessableEvent[] = [];
+
+    for(let i = 0; i < 5; i++) {
+      const id = `id-${i}`;
+      const event = createImpressionEvent(id);
+      events.push(event);
+      await cache.set(id, { id, event });
+    }
+
+    const processor = new ReactNativeNetInfoEventProcessor({
+      eventDispatcher,
+      dispatchRepeater,
+      failedEventRepeater,
+      batchSize: 1000,
+      eventStore: cache,
+    });
+
+    processor.start();
+    await processor.onRunning();
+
+    mockNetInfo.pushState(true);
+    expect(eventDispatcher.dispatchEvent).not.toHaveBeenCalled();
+
+    mockNetInfo.pushState(true);
+    expect(eventDispatcher.dispatchEvent).not.toHaveBeenCalled();
+  });
+
+  it('should retry failed events when network becomes reachable', async () => {
+    const eventDispatcher = getMockDispatcher();
+    const dispatchRepeater = getMockRepeater();
+    const failedEventRepeater = getMockRepeater();
+
+    const cache = getMockAsyncCache<EventWithId>();
+    const events: ProcessableEvent[] = [];
+
+    for(let i = 0; i < 5; i++) {
+      const id = `id-${i}`;
+      const event = createImpressionEvent(id);
+      events.push(event);
+      await cache.set(id, { id, event });
+    }
+
+    const processor = new ReactNativeNetInfoEventProcessor({
+      eventDispatcher,
+      dispatchRepeater,
+      failedEventRepeater,
+      batchSize: 1000,
+      eventStore: cache,
+    });
+
+    processor.start();
+    await processor.onRunning();
+
+    mockNetInfo.pushState(false);
+    expect(eventDispatcher.dispatchEvent).not.toHaveBeenCalled();
+
+    mockNetInfo.pushState(true);
+
+    await exhaustMicrotasks();
+
+    expect(eventDispatcher.dispatchEvent).toHaveBeenCalledWith(formatEvents(events));
+  });
+
+  it('should unsubscribe from netinfo listener when stopped', async () => {
+    const eventDispatcher = getMockDispatcher();
+    const dispatchRepeater = getMockRepeater();
+    const failedEventRepeater = getMockRepeater();
+
+    const cache = getMockAsyncCache<EventWithId>();
+
+    const processor = new ReactNativeNetInfoEventProcessor({
+      eventDispatcher,
+      dispatchRepeater,
+      failedEventRepeater,
+      batchSize: 1000,
+      eventStore: cache,
+    });
+
+    processor.start();
+    await processor.onRunning();
+
+    mockNetInfo.pushState(false);
+
+    processor.stop();
+    await processor.onTerminated();
+
+    expect(mockNetInfo.unsubs[0]).toHaveBeenCalled();
+  });
+});

lib/event_processor/batch_event_processor.react_native.ts

@@ -0,0 +1,40 @@
+import {
+  NetInfoState,
+  addEventListener as addConnectionListener,
+} from '@react-native-community/netinfo';
+
+import { BatchEventProcessor, BatchEventProcessorConfig } from './batch_event_processor';
+import { Fn } from '../utils/type';
+
+export class ReactNativeNetInfoEventProcessor extends BatchEventProcessor {
+  private isInternetReachable = true;
+  private unsubscribeNetInfo?: Fn;
+
+  constructor(config: BatchEventProcessorConfig) {
+    super(config);
+  }
+
+  private async connectionListener(state: NetInfoState) {
+    if (this.isInternetReachable && !state.isInternetReachable) {
+      this.isInternetReachable = false;
+      return;
+    }
+
+    if (!this.isInternetReachable && state.isInternetReachable) {
+      this.isInternetReachable = true;
+      this.retryFailedEvents();
+    }
+  }
+
+  start(): void {
+    super.start();
+    this.unsubscribeNetInfo = addConnectionListener(this.connectionListener.bind(this));
+  }
+
+  stop(): void {
+    if (this.unsubscribeNetInfo) {
+      this.unsubscribeNetInfo();
+    }
+    super.stop();
+  }
+}