feat: add queuing to event-clients

Author: AlemTuzlak

examples/react/start/src/plugin.ts

@@ -19,8 +19,14 @@ class QueryDevtoolsClient extends EventClient<EventMap> {
   constructor() {
     super({
       pluginId: 'query-devtools',
+      debug: true,
     })
   }
 }
 
 export const queryPlugin = new QueryDevtoolsClient()
+// this should be queued and emitted when bus is available
+queryPlugin.emit('test', {
+  title: 'Query Devtools',
+  description: 'A plugin for query debugging',
+})

packages/event-bus-client/src/plugin.ts

@@ -23,7 +23,26 @@ export class EventClient<
   #pluginId: TPluginId
   #eventTarget: () => EventTarget
   #debug: boolean
+  #queuedEvents: Array<TanStackDevtoolsEvent<string, any>>
+  #connected: boolean
+  #connectIntervalId: number | null
+  #connectEveryMs: number
+  #retryCount = 0
+  #maxRetries = 5
 
+  #connectFunction = () => {
+    if (this.#retryCount < this.#maxRetries) {
+      this.#retryCount++
+      this.#eventTarget().dispatchEvent(new CustomEvent('tanstack-connect'))
+      return
+    }
+    this.#eventTarget().removeEventListener(
+      'tanstack-connect',
+      this.#connectFunction,
+    )
+    this.debugLog('Max retries reached, giving up on connection')
+    clearInterval(this.#connectIntervalId!)
+  }
   constructor({
     pluginId,
     debug = false,
@@ -35,6 +54,46 @@ export class EventClient<
     this.#eventTarget = this.getGlobalTarget
     this.#debug = debug
     this.debugLog(' Initializing event subscription for plugin', this.#pluginId)
+    this.#queuedEvents = []
+    this.#connected = false
+    this.#connectIntervalId = null
+    this.#connectEveryMs = 600
+    this.startConnectLoop()
+  }
+
+  private startConnectLoop() {
+    if (this.#connectIntervalId !== null) return
+    this.debugLog(`Starting connect loop (every ${this.#connectEveryMs}ms)`)
+    const onConnected = () => {
+      this.debugLog('Connected to event bus')
+      this.#connected = true
+      this.debugLog('Emitting queued events', this.#queuedEvents)
+      this.#queuedEvents.forEach((event) => this.emitEventToBus(event))
+      this.#queuedEvents = []
+      this.stopConnectLoop()
+      this.#eventTarget().removeEventListener(
+        'tanstack-connect-success',
+        onConnected,
+      )
+    }
+    this.#eventTarget().addEventListener(
+      'tanstack-connect-success',
+      onConnected,
+    )
+
+    this.#connectIntervalId = setInterval(
+      this.#connectFunction,
+      this.#connectEveryMs,
+    ) as unknown as number
+  }
+
+  private stopConnectLoop() {
+    if (this.#connectIntervalId === null) {
+      return
+    }
+    clearInterval(this.#connectIntervalId)
+    this.#connectIntervalId = null
+    this.debugLog('Stopped connect loop')
   }
 
   private debugLog(...args: Array<any>) {
@@ -84,7 +143,17 @@ export class EventClient<
     eventSuffix: TSuffix,
     payload: TEventMap[`${TPluginId & string}:${TSuffix}`],
   ) {
-    this.emitEventToBus({
+    // wait to connect to the bus
+    if (!this.#connected) {
+      this.debugLog('Bus not available, will be pushed as soon as connected')
+      return this.#queuedEvents.push({
+        type: `${this.#pluginId}:${eventSuffix}`,
+        payload,
+        pluginId: this.#pluginId,
+      })
+    }
+    // emit right now
+    return this.emitEventToBus({
       type: `${this.#pluginId}:${eventSuffix}`,
       payload,
       pluginId: this.#pluginId,

packages/event-bus/src/client/client.ts

@@ -30,11 +30,18 @@ export class ClientEventBus {
   #eventTarget: EventTarget
   #debug: boolean
   #connectToServerBus: boolean
+
   #dispatcher = (e: Event) => {
     const event = (e as CustomEvent).detail
     this.emitToServer(event)
     this.emitToClients(event)
   }
+  #connectFunction = () => {
+    this.debugLog(
+      'Connection request made to event-bus, replying back with success',
+    )
+    this.#eventTarget.dispatchEvent(new CustomEvent('tanstack-connect-success'))
+  }
   constructor({
     port = 42069,
     debug = false,
@@ -46,6 +53,7 @@ export class ClientEventBus {
     this.#socket = null
     this.#connectToServerBus = connectToServerBus
     this.#eventTarget = this.getGlobalTarget()
+
     this.debugLog('Initializing client event bus')
   }
 
@@ -91,6 +99,10 @@ export class ClientEventBus {
       'tanstack-dispatch-event',
       this.#dispatcher,
     )
+    this.#eventTarget.addEventListener(
+      'tanstack-connect',
+      this.#connectFunction,
+    )
   }
   stop() {
     this.debugLog('Stopping client event bus')
@@ -101,6 +113,10 @@ export class ClientEventBus {
       'tanstack-dispatch-event',
       this.#dispatcher,
     )
+    this.#eventTarget.removeEventListener(
+      'tanstack-connect',
+      this.#connectFunction,
+    )
     this.#eventSource?.close()
     this.#socket?.close()
     this.#socket = null

packages/event-bus/src/server/server.ts

@@ -33,7 +33,9 @@ export class ServerEventBus {
     this.debugLog('Dispatching event from dispatcher, forwarding', event)
     this.emit(event)
   }
-
+  #connectFunction = () => {
+    this.#eventTarget.dispatchEvent(new CustomEvent('tanstack-connect-success'))
+  }
   constructor({ port = 42069, debug = false } = {}) {
     this.#port = port
     this.#eventTarget = globalThis.__EVENT_TARGET__ ?? new EventTarget()
@@ -165,6 +167,10 @@ export class ServerEventBus {
       'tanstack-dispatch-event',
       this.#dispatcher,
     )
+    this.#eventTarget.addEventListener(
+      'tanstack-connect',
+      this.#connectFunction,
+    )
     this.handleNewConnection(wss)
 
     // Handle connection upgrade for WebSocket
@@ -200,6 +206,10 @@ export class ServerEventBus {
       'tanstack-dispatch-event',
       this.#dispatcher,
     )
+    this.#eventTarget.removeEventListener(
+      'tanstack-connect',
+      this.#connectFunction,
+    )
     this.debugLog('[tanstack-devtools] All connections cleared')
   }
 }