feat: Emit state change events from FluxConnection (#405) (#434)

* feat: Emit state change events from FluxConnection

These can be used to implement reconnect logic for Fluxes but also to update the UI to indicate that server pushes are not being received

* Use EventTarget

* Extend EventTarget

* Merge with an interface

Co-authored-by: Anton Platonov <[email protected]>
(cherry picked from commit 134dbb5)

Co-authored-by: Artur <[email protected]>

packages/ts/hilla-frontend/mocks/socket.io-client.ts

@@ -2,15 +2,20 @@ interface Socket {}
 
 export const io = (path: string, options: any): Socket => {
   const sentMessages = [];
-  const incomingMessages = [];
+  const eventHandlers = {};
   return {
-    on: (type: string, event: any) => {
-      incomingMessages.push({ type, event });
+    on: (event: string, listener: any) => {
+      if (!eventHandlers[event]) {
+        eventHandlers[event] = [];
+      }
+      eventHandlers[event].push(listener);
+    },
+    emit: (event: string, ...args: any[]) => {
+      (eventHandlers[event] || []).forEach((l: any) => l(args));
     },
     send: (...args: any[]) => {
       sentMessages.push(...args);
     },
     sentMessages,
-    incomingMessages,
   };
 };

packages/ts/hilla-frontend/src/Connect.ts

@@ -249,7 +249,7 @@ function isFlowLoaded(): boolean {
 }
 
 /**
- * Hilla Connect client class is a low-level network calling utility. It stores
+ * A low-level network calling utility. It stores
  * a prefix and facilitates remote calls to endpoint class methods
  * on the Hilla backend.
  *
@@ -281,7 +281,7 @@ export class ConnectClient {
    */
   public middlewares: Middleware[] = [];
 
-  private fluxConnection: FluxConnection | undefined = undefined;
+  private _fluxConnection: FluxConnection | undefined = undefined;
 
   /**
    * @param options Constructor options.
@@ -426,10 +426,16 @@ export class ConnectClient {
    * @returns {} A subscription used to handles values as they become available.
    */
   public subscribe(endpoint: string, method: string, params?: any): Subscription<any> {
-    if (!this.fluxConnection) {
-      this.fluxConnection = new FluxConnection();
-    }
-
     return this.fluxConnection.subscribe(endpoint, method, params ? Object.values(params) : []);
   }
+
+  /**
+   * Gets a representation of the underlying persistent network connection used for subscribing to Flux type endpoint methods.
+   */
+  get fluxConnection(): FluxConnection {
+    if (!this._fluxConnection) {
+      this._fluxConnection = new FluxConnection();
+    }
+    return this._fluxConnection;
+  }
 }

packages/ts/hilla-frontend/src/FluxConnection.ts

@@ -5,16 +5,38 @@ import type { Subscription } from './Connect';
 import { getCsrfTokenHeadersForEndpointRequest } from './CsrfUtils';
 import type { ClientMessage, ServerCloseMessage, ServerConnectMessage, ServerMessage } from './FluxMessages';
 
-export class FluxConnection {
+export enum State {
+  ACTIVE = 'active',
+  INACTIVE = 'inactive',
+}
+
+type ActiveEvent = CustomEvent<{ active: boolean }>;
+interface EventMap {
+  'state-changed': ActiveEvent;
+}
+
+type ListenerType<T extends keyof EventMap> =
+  | ((this: FluxConnection, ev: EventMap[T]) => any)
+  | {
+      handleEvent(ev: EventMap[T]): void;
+    }
+  | null;
+
+/**
+ * A representation of the underlying persistent network connection used for subscribing to Flux type endpoint methods.
+ */
+export class FluxConnection extends EventTarget {
   private nextId = 0;
   private endpointInfos = new Map<string, string>();
   private onNextCallbacks = new Map<string, (value: any) => void>();
   private onCompleteCallbacks = new Map<string, () => void>();
   private onErrorCallbacks = new Map<string, () => void>();
 
   private socket!: Socket<DefaultEventsMap, DefaultEventsMap>;
+  public state: State = State.INACTIVE;
 
   constructor() {
+    super();
     if (!(window as any).Vaadin?.featureFlags?.hillaPush) {
       // Remove when removing feature flag
       throw new Error(
@@ -30,6 +52,24 @@ export class FluxConnection {
     this.socket.on('message', (message) => {
       this.handleMessage(JSON.parse(message));
     });
+    this.socket.on('disconnect', () => {
+      // https://socket.io/docs/v4/client-api/#event-disconnect
+      if (this.state === State.ACTIVE) {
+        this.state = State.INACTIVE;
+        this.dispatchEvent(new CustomEvent('state-changed', { detail: { active: false } }));
+      }
+    });
+    this.socket.on('connect_error', () => {
+      // https://socket.io/docs/v4/client-api/#event-connect_error
+    });
+
+    this.socket.on('connect', () => {
+      // https://socket.io/docs/v4/client-api/#event-connect
+      if (this.state === State.INACTIVE) {
+        this.state = State.ACTIVE;
+        this.dispatchEvent(new CustomEvent('state-changed', { detail: { active: true } }));
+      }
+    });
   }
 
   private handleMessage(message: ClientMessage) {
@@ -73,10 +113,18 @@ export class FluxConnection {
     this.socket.send(message);
   }
 
-  subscribe(endpointName: string, methodName: string, maybeParams?: Array<any>): Subscription<any> {
+  /**
+   * Subscribes to the flux returned by the given endpoint name + method name using the given parameters.
+   *
+   * @param endpointName the endpoint to connect to
+   * @param methodName the method in the endpoint to connect to
+   * @param parameters the parameters to use
+   * @returns a subscription
+   */
+  subscribe(endpointName: string, methodName: string, parameters?: Array<any>): Subscription<any> {
     const id: string = this.nextId.toString();
     this.nextId += 1;
-    const params = maybeParams || [];
+    const params = parameters || [];
 
     const msg: ServerConnectMessage = { '@type': 'subscribe', id, endpointName, methodName, params };
     const endpointInfo = `${endpointName}.${methodName}(${JSON.stringify(params)})`;
@@ -117,3 +165,8 @@ export class FluxConnection {
     return hillaSubscription;
   }
 }
+
+export interface FluxConnection {
+  addEventListener<T extends keyof EventMap>(type: T, listener: ListenerType<T>): void;
+  removeEventListener<T extends keyof EventMap>(type: T, listener: ListenerType<T>): void;
+}

packages/ts/hilla-frontend/src/index.ts

@@ -1,6 +1,6 @@
 export * from './Authentication.js';
 export * from './Connect.js';
-export { FluxConnection } from './FluxConnection';
+export { FluxConnection, State } from './FluxConnection';
 
 const $wnd = window as any;
 /* c8 ignore next 2 */

packages/ts/hilla-frontend/test/Connect.test.ts

@@ -550,22 +550,22 @@ describe('ConnectClient', () => {
     });
 
     it('should create a fluxConnection', async () => {
-      (client as any).fluxConnection = undefined; // NOSONAR
+      (client as any)._fluxConnection = undefined; // NOSONAR
       client.subscribe('FooEndpoint', 'fooMethod');
-      expect((client as any).fluxConnection).to.not.equal(undefined);
+      expect((client as any)._fluxConnection).to.not.equal(undefined);
     });
 
     it('should reuse the fluxConnection', async () => {
       client.subscribe('FooEndpoint', 'fooMethod');
       const { fluxConnection } = client as any;
       client.subscribe('FooEndpoint', 'barMethod');
-      expect((client as any).fluxConnection).to.equal(fluxConnection);
+      expect((client as any)._fluxConnection).to.equal(fluxConnection);
     });
 
     it('should call FluxConnection', async () => {
-      (client as any).fluxConnection = new FluxConnection();
+      (client as any)._fluxConnection = new FluxConnection();
       let called = 0;
-      (client as any).fluxConnection.subscribe = (endpointName: any, methodName: any, params: any) => {
+      (client as any)._fluxConnection.subscribe = (endpointName: any, methodName: any, params: any) => {
         called += 1;
         expect(endpointName).to.equal('FooEndpoint');
         expect(methodName).to.equal('fooMethod');

packages/ts/hilla-frontend/test/FluxConnection.test.ts

@@ -256,4 +256,48 @@ describe('FluxConnection', () => {
     fakeElement.disconnectedCallback();
     expectNoDataRetained(fluxConnectionAny);
   });
+  it('dispatches an active event on socket.io connect', () => {
+    const { socket } = fluxConnectionAny;
+    socket.connected = false;
+    let events = 0;
+    fluxConnection.addEventListener('state-changed', (e) => {
+      if (e.detail.active) {
+        events += 1;
+      }
+    });
+    socket.connected = true;
+    socket.emit('connect');
+    expect(events).to.equal(1);
+  });
+  it('dispatches an active event on socket.io reconnect', () => {
+    const { socket } = fluxConnectionAny;
+    socket.connected = false;
+    let events = 0;
+    fluxConnection.addEventListener('state-changed', (e) => {
+      if (e.detail.active) {
+        events += 1;
+      }
+    });
+    socket.connected = true;
+    socket.emit('connect');
+    socket.connected = false;
+    socket.emit('disconnect');
+    socket.connected = true;
+    socket.emit('connect');
+    expect(events).to.equal(2);
+  });
+  it('dispatches an inactive event on socket.io disconnect', () => {
+    const { socket } = fluxConnectionAny;
+    let events = 0;
+    fluxConnection.addEventListener('state-changed', (e) => {
+      if (!e.detail.active) {
+        events += 1;
+      }
+    });
+    socket.connected = true;
+    socket.emit('connect');
+    socket.connected = false;
+    socket.emit('disconnect');
+    expect(events).to.equal(1);
+  });
 });