feat: waiting plugin

add waiting plugin interface.
any plugin implement this interface will pause event processing when it's added to analytics
any plugin implement this interface can pause and resume event processing as needed

Author: Sunita Prajapati

examples/AnalyticsReactNativeExample/plugins/ExampleWaitingPlugin.tsx

@@ -0,0 +1,33 @@
+import {
+  WaitingPlugin,
+  PluginType,
+  Plugin,
+  
+} from '@segment/analytics-react-native';
+
+import type {SegmentAPISettings, SegmentClient, SegmentEvent, UpdateType} from '@segment/analytics-react-native';
+export class ExampleWaitingPlugin extends WaitingPlugin {
+  type = PluginType.enrichment;
+  analytics = undefined;
+  tracked = false;
+
+  /**
+   * Called when settings are updated
+   */
+  update(_settings: SegmentAPISettings, _type: UpdateType) {
+    if (this.type === PluginType.before) {
+      // delay 3 seconds, then resume event processing
+      setTimeout(() => {
+        this.resume();   
+      }, 3000);
+    }
+  }
+
+  /**
+   * Called for track events
+   */
+  track(event: SegmentEvent) {
+    this.tracked = true;
+    return event;  
+  }
+}

packages/core/src/analytics.ts

@@ -72,7 +72,7 @@ import {
   translateHTTPError,
 } from './errors';
 import { QueueFlushingPlugin } from './plugins/QueueFlushingPlugin';
-import { WaitingPlugin } from './plugins/Waiting';
+import { WaitingPlugin } from './plugin';
 
 type OnPluginAddedCallback = (plugin: Plugin) => void;
 
@@ -98,11 +98,6 @@ export class SegmentClient {
   private isAddingPlugins = false;
 
   private timeline: Timeline;
-  // running state (matches Kotlin's running flag)
-  private isRunning = true;
-
-  // Waiting plugin instance (buffers events while paused)
-  private waitingPlugin?: WaitingPlugin;
 
   private pluginsToAdd: Plugin[] = [];
 
@@ -206,17 +201,17 @@ export class SegmentClient {
     this.store = store;
     this.timeline = new Timeline();
 
-    // create and add waiting plugin immediately so early events get buffered.
-    try {
-      this.waitingPlugin = new WaitingPlugin();
-      // add directly to timeline via addPlugin to ensure configure() is called immediately
-      this.addPlugin(this.waitingPlugin);
-      // initial running state false until init completes (mirrors Kotlin semantics)
-      this.isRunning = false;
-    } catch (e) {
-      // if WaitingPlugin instantiation or add fails, fallback to running=true
-      this.isRunning = true;
-    }
+    // // create and add waiting plugin immediately so early events get buffered.
+    // try {
+    //   this.waitingPlugin = new WaitingPlugin();
+    //   // add directly to timeline via addPlugin to ensure configure() is called immediately
+    //   this.addPlugin(this.waitingPlugin);
+    //   // initial running state false until init completes (mirrors Kotlin semantics)
+    //   this.isRunning = false;
+    // } catch (e) {
+    //   // if WaitingPlugin instantiation or add fails, fallback to running=true
+    //   this.isRunning = true;
+    // }
 
     // Initialize the watchables
     this.context = {
@@ -313,10 +308,6 @@ export class SegmentClient {
       if ((await this.store.isReady.get(true)) === false) {
         await this.storageReady();
       }
-
-      // Pause pipeline at init start (buffer events until init completes)
-      this.pauseEventProcessing();
-
       // Get new settings from segment
       // It's important to run this before checkInstalledVersion and trackDeeplinks to give time for destination plugins
       // which make use of the settings object to initialize
@@ -330,8 +321,6 @@ export class SegmentClient {
       ]);
       await this.onReady();
       this.isReady.value = true;
-      // Resume pipeline before processing pending events so WaitingPlugin flushes
-      await this.resumeEventProcessing();
       // Process all pending events
       await this.processPendingEvents();
       // Trigger manual flush
@@ -487,17 +476,24 @@ export class SegmentClient {
         settings
       );
     }
-
-    if (!this.isReady.value) {
+    console.log('!this.isReady.value', !this.isReady.value);
+    if (!this.isReady.value && !(plugin instanceof WaitingPlugin)) {
       this.pluginsToAdd.push(plugin);
     } else {
+      console.log('this.addPlugin');
       this.addPlugin(plugin);
     }
   }
 
   private addPlugin(plugin: Plugin) {
     plugin.configure(this);
     this.timeline.add(plugin);
+    //check for waiting plugin here
+    if (plugin instanceof WaitingPlugin) {
+      console.log('add plugin');
+      this.pauseEventProcessingForPlugin(plugin);
+    }
+
     this.triggerOnPluginLoaded(plugin);
   }
 
@@ -534,7 +530,7 @@ export class SegmentClient {
   ): Promise<SegmentEvent | undefined> {
     const event = await this.applyContextData(incomingEvent);
     this.flushPolicyExecuter.notify(event);
-    return this.timeline.process(event);
+    return await this.timeline.process(event);
   }
 
   private async trackDeepLinks() {
@@ -1049,45 +1045,58 @@ export class SegmentClient {
 
     return totalEventsCount;
   }
-  /*
-   * Running / pause/resume helpers (Kotlin parity)
-   */
+  private resumeTimeoutId?: ReturnType<typeof setTimeout>;
+  private waitingPlugins = new Set<WaitingPlugin>();
 
-  public running() {
-    return this.isRunning;
-  }
   /**
    * Pause event processing globally. Events will be buffered into pendingEvents and WaitingPlugin.
    * An auto-resume will be scheduled after `timeout` ms.
    */
-  public pauseEventProcessing(timeout = 30000) {
-    if (!this.isRunning) {
+  pauseEventProcessingForPlugin(plugin?: WaitingPlugin) {
+    if (plugin) {
+      this.waitingPlugins.add(plugin);
+    }
+    this.pauseEventProcessing();
+  }
+  async resumeEventProcessingForPlugin(plugin?: WaitingPlugin) {
+    if (plugin) {
+      this.waitingPlugins.delete(plugin);
+    }
+    if (this.waitingPlugins.size > 0) {
+      return; // still blocked
+    }
+
+    await this.resumeEventProcessing();
+  }
+
+  pauseEventProcessing(timeout = 30000) {
+    // IMPORTANT: ignore repeated pauses
+    if (!this.isReady.value) {
       return;
     }
 
-    this.isRunning = false;
-    try {
-      this.waitingPlugin?.pause();
-    } catch {
-      // ignore if plugin not present
+    this.isReady.value = false;
+
+    // clear previous timeout if any
+    if (this.resumeTimeoutId) {
+      clearTimeout(this.resumeTimeoutId);
     }
 
-    // auto-resume after timeout to avoid permanent blocking
-    setTimeout(() => {
-      void this.resumeEventProcessing();
+    this.resumeTimeoutId = setTimeout(async () => {
+      await this.resumeEventProcessing();
     }, timeout);
   }
-  public async resumeEventProcessing() {
-    if (this.isRunning) {
+  async resumeEventProcessing() {
+    if (this.isReady.value) {
       return;
     }
 
-    this.isRunning = true;
-
-    try {
-      await this.waitingPlugin?.resume();
-    } catch {
-      // ignore plugin errors during resume
+    if (this.resumeTimeoutId) {
+      clearTimeout(this.resumeTimeoutId);
+      this.resumeTimeoutId = undefined;
     }
+    // this.waitingPlugins.clear();
+    this.isReady.value = true;
+    await this.processPendingEvents();
   }
 }

packages/core/src/plugin.ts

@@ -115,12 +115,14 @@ export class DestinationPlugin extends EventPlugin {
   key = '';
 
   timeline = new Timeline();
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  store: any;
 
   private hasSettings() {
     return this.analytics?.settings.get()?.[this.key] !== undefined;
   }
 
-  private isEnabled(event: SegmentEvent): boolean {
+  protected isEnabled(event: SegmentEvent): boolean {
     let customerDisabled = false;
     if (event.integrations?.[this.key] === false) {
       customerDisabled = true;
@@ -140,6 +142,10 @@ export class DestinationPlugin extends EventPlugin {
     if (analytics) {
       plugin.configure(analytics);
     }
+
+    if (analytics && plugin instanceof WaitingPlugin) {
+      analytics.pauseEventProcessingForPlugin(plugin);
+    }
     this.timeline.add(plugin);
     return plugin;
   }
@@ -179,7 +185,6 @@ export class DestinationPlugin extends EventPlugin {
       type: PluginType.before,
       event,
     });
-
     if (beforeResult === undefined) {
       return;
     }
@@ -210,3 +215,27 @@ export class UtilityPlugin extends EventPlugin {}
 
 // For internal platform-specific bits
 export class PlatformPlugin extends Plugin {}
+
+export { PluginType };
+
+/**
+ * WaitingPlugin
+ * Buffers events when paused and releases them when resumed.
+ */
+export class WaitingPlugin extends Plugin {
+  constructor() {
+    super();
+  }
+
+  configure(analytics: SegmentClient) {
+    super.configure(analytics);
+  }
+
+  pause() {
+    this.analytics?.pauseEventProcessingForPlugin(this);
+  }
+
+  async resume() {
+    await this.analytics?.resumeEventProcessingForPlugin(this);
+  }
+}