Enhancement clear queue flush (#1071)

* feat: expose methods to clear queue and get event count

* feat: added test cases for clear queue and get count

* refactor: refactored method names and some logic

* fix: fixed test cases

* feat: changed the logic to touch minimum file changes

Author: sunitaprajapati89

packages/core/src/analytics.ts

@@ -71,6 +71,7 @@ import {
   SegmentError,
   translateHTTPError,
 } from './errors';
+import { QueueFlushingPlugin } from './plugins/QueueFlushingPlugin';
 
 type OnPluginAddedCallback = (plugin: Plugin) => void;
 
@@ -982,4 +983,46 @@ export class SegmentClient {
       userId: userInfo.userId,
     };
   };
+  /* Method for clearing flush queue */
+  clear() {
+    const plugins = this.getPlugins();
+
+    plugins.forEach(async (plugin) => {
+      if (plugin instanceof SegmentDestination) {
+        const timelinePlugins = plugin.timeline?.plugins?.after ?? [];
+
+        for (const subPlugin of timelinePlugins) {
+          if (subPlugin instanceof QueueFlushingPlugin) {
+            await subPlugin.dequeueEvents();
+          }
+        }
+      }
+    });
+
+    this.flushPolicyExecuter.reset();
+  }
+
+  /**
+   * Method to get count of events in flush queue.
+   */
+  async pendingEvents() {
+    const plugins = this.getPlugins();
+    let totalEventsCount = 0;
+
+    for (const plugin of plugins) {
+      // We're looking inside SegmentDestination's `after` plugins
+      if (plugin instanceof SegmentDestination) {
+        const timelinePlugins = plugin.timeline?.plugins?.after ?? [];
+
+        for (const subPlugin of timelinePlugins) {
+          if (subPlugin instanceof QueueFlushingPlugin) {
+            const eventsCount = await subPlugin.pendingEvents();
+            totalEventsCount += eventsCount;
+          }
+        }
+      }
+    }
+
+    return totalEventsCount;
+  }
 }

packages/core/src/plugins/QueueFlushingPlugin.ts

@@ -130,4 +130,20 @@ export class QueueFlushingPlugin extends UtilityPlugin {
       return { events: filteredEvents };
     });
   }
+  /**
+   * Clear all events from the queue
+   */
+  async dequeueEvents() {
+    await this.queueStore?.dispatch(() => {
+      return { events: [] };
+    });
+  }
+
+  /**
+   * * Returns the count of items in the queue
+   */
+  async pendingEvents() {
+    const events = (await this.queueStore?.getState(true))?.events ?? [];
+    return events.length;
+  }
 }

packages/core/src/plugins/__tests__/QueueFlushingPlugin.test.ts

@@ -77,4 +77,67 @@ describe('QueueFlushingPlugin', () => {
     // @ts-ignore
     expect(queuePlugin.queueStore?.getState().events).toHaveLength(0);
   });
+  it('should clear all events from the queue', async () => {
+    const onFlush = jest.fn().mockResolvedValue(undefined);
+    const queuePlugin = setupQueuePlugin(onFlush, 10);
+    const event1: SegmentEvent = {
+      type: EventType.TrackEvent,
+      event: 'test1',
+      properties: {
+        test: 'test1',
+      },
+    };
+    const event2: SegmentEvent = {
+      type: EventType.TrackEvent,
+      event: 'test2',
+      properties: {
+        test: 'test2',
+      },
+    };
+    await queuePlugin.execute(event1);
+    await queuePlugin.execute(event2);
+    // eslint-disable-next-line @typescript-eslint/ban-ts-comment
+    // @ts-ignore
+    expect(queuePlugin.queueStore?.getState().events).toHaveLength(2);
+    await queuePlugin.dequeueEvents();
+    // eslint-disable-next-line @typescript-eslint/ban-ts-comment
+    // @ts-ignore
+    expect(queuePlugin.queueStore?.getState().events).toHaveLength(0);
+  });
+  it('should return the count of items in the queue', async () => {
+    const onFlush = jest.fn().mockResolvedValue(undefined);
+    const queuePlugin = setupQueuePlugin(onFlush, 10);
+
+    const event1: SegmentEvent = {
+      type: EventType.TrackEvent,
+      event: 'test1',
+      properties: {
+        test: 'test1',
+      },
+    };
+
+    const event2: SegmentEvent = {
+      type: EventType.TrackEvent,
+      event: 'test2',
+      properties: {
+        test: 'test2',
+      },
+    };
+
+    await queuePlugin.execute(event1);
+    await queuePlugin.execute(event2);
+
+    let eventsCount = await queuePlugin.pendingEvents();
+    expect(eventsCount).toBe(2);
+
+    await queuePlugin.dequeue(event1);
+
+    eventsCount = await queuePlugin.pendingEvents();
+    expect(eventsCount).toBe(1);
+
+    await queuePlugin.dequeueEvents();
+
+    eventsCount = await queuePlugin.pendingEvents();
+    expect(eventsCount).toBe(0);
+  });
 });

packages/core/src/storage/types.ts

@@ -92,7 +92,6 @@ export interface Storage {
   readonly pendingEvents: Watchable<SegmentEvent[]> &
     Settable<SegmentEvent[]> &
     Queue<SegmentEvent, SegmentEvent[]>;
-    
   readonly enabled: Watchable<boolean> & Settable<boolean>;
 }
 export type DeepLinkData = {