Zg/resume on reconnect

src/Client.ts

@@ -1,4 +1,4 @@
-/** THIS FILE IS MANUALLY MAINAINED: see .fernignore */
+/** THIS FILE IS MANUALLY MAINTAINED: see .fernignore */
 
 import * as environments from "./environments";
 import * as core from "./core";

src/api/resources/empathicVoice/client/Client.ts

@@ -1,4 +1,4 @@
-/** THIS FILE IS MANUALLY MAINAINED: see .fernignore */
+/** THIS FILE IS MANUALLY MAINTAINED: see .fernignore */
 
 import * as environments from "../../../../environments";
 import * as core from "../../../../core";

src/api/resources/empathicVoice/resources/chat/client/Client.ts

@@ -1,4 +1,4 @@
-/** THIS FILE IS MANUALLY MAINAINED: see .fernignore */
+/** THIS FILE IS MANUALLY MAINTAINED: see .fernignore */
 import * as environments from "../../../../../../environments";
 import * as core from "../../../../../../core";
 import qs from "qs";
@@ -33,6 +33,9 @@ export declare namespace Chat {
 
         /** Extra query parameters sent at WebSocket connection */
         queryParams?: Record<string, string | string[] | object | object[]>;
+
+        /** Enable resuming the Chat on specific disconnects. If `true`, the SDK will attempt to reconnect using the `chat_group_id` if a disconnect occurs with [close codes](https://developer.mozilla.org/en-US/docs/Web/API/CloseEvent/code#value): `1006`, `1011`, `1012`, `1013`, or `1014`. Defaults to `true`. */
+        shouldResumeChat?: boolean;
     }
 }
 
@@ -77,6 +80,7 @@ export class Chat {
             }
         }
 
+        const shouldResumeChat = args.shouldResumeChat ?? true;
         const socket = new core.ReconnectingWebSocket(
             `wss://${(core.Supplier.get(this._options.environment) ?? environments.HumeEnvironment.Production).replace(
                 "https://",
@@ -86,11 +90,20 @@ export class Chat {
             {
                 debug: args.debug ?? false,
                 maxRetries: args.reconnectAttempts ?? 30,
+                shouldAttemptReconnectHook: (event) => Chat._staticShouldAttemptReconnectEvi(event, shouldResumeChat),
             },
         );
 
-        return new ChatSocket({
-            socket,
-        });
+        return new ChatSocket({ socket, shouldResumeChat });
+    }
+
+    private static _staticShouldAttemptReconnectEvi(event: core.CloseEvent, shouldResume: boolean): boolean {
+        // Defer to default logic
+        if (!shouldResume) return true;
+        // Allow attempt
+        const resumableCloseCodes: Set<number> = new Set([1006, 1011, 1012, 1013, 1014]);
+        if (resumableCloseCodes.has(event.code)) return true;
+        // Prevent attempt
+        return false;
     }
 }

src/api/resources/empathicVoice/resources/chat/client/Socket.ts

@@ -1,4 +1,4 @@
-/** THIS FILE IS MANUALLY MAINAINED: see .fernignore */
+/** THIS FILE IS MANUALLY MAINTAINED: see .fernignore */
 import * as core from "../../../../../../core";
 import * as errors from "../../../../../../errors";
 import * as Hume from "../../../../../index";
@@ -7,6 +7,7 @@ import * as serializers from "../../../../../../serialization/index";
 export declare namespace ChatSocket {
     interface Args {
         socket: core.ReconnectingWebSocket;
+        shouldResumeChat?: boolean;
     }
 
     type Response = Hume.empathicVoice.SubscribeEvent & { receivedAt: Date };
@@ -24,8 +25,11 @@ export class ChatSocket {
 
     protected readonly eventHandlers: ChatSocket.EventHandlers = {};
 
-    constructor({ socket }: ChatSocket.Args) {
+    private readonly _shouldResumeChat: boolean;
+
+    constructor({ socket, shouldResumeChat }: ChatSocket.Args) {
         this.socket = socket;
+        this._shouldResumeChat = shouldResumeChat ?? true;
 
         this.socket.addEventListener("open", this.handleOpen);
         this.socket.addEventListener("message", this.handleMessage);
@@ -220,10 +224,11 @@ export class ChatSocket {
             breadcrumbsPrefix: ["response"],
         });
         if (parsedResponse.ok) {
-            this.eventHandlers.message?.({
-                ...parsedResponse.value,
-                receivedAt: new Date(),
-            });
+            const message = parsedResponse.value;
+            if (message.type === "chat_metadata" && this._shouldResumeChat) {
+                this.socket.setQueryParamOverride("resumed_chat_group_id", message.chatGroupId);
+            }
+            this.eventHandlers.message?.({ ...message, receivedAt: new Date() });
         } else {
             this.eventHandlers.error?.(new Error(`Received unknown message type`));
         }

src/api/resources/empathicVoice/resources/chat/client/index.ts

@@ -1,3 +1,3 @@
-/** THIS FILE IS MANUALLY MAINAINED: see .fernignore */
+/** THIS FILE IS MANUALLY MAINTAINED: see .fernignore */
 export { ChatSocket } from "./Socket";
 export { Chat } from "./Client";

src/api/resources/empathicVoice/resources/chat/index.ts

@@ -1,3 +1,3 @@
-/** THIS FILE IS MANUALLY MAINAINED: see .fernignore */
+/** THIS FILE IS MANUALLY MAINTAINED: see .fernignore */
 export * from "./types";
 export * from "./client";

src/core/fetcher/Supplier.ts

@@ -1,4 +1,4 @@
-/** THIS FILE IS MANUALLY MAINAINED: see .fernignore */
+/** THIS FILE IS MANUALLY MAINTAINED: see .fernignore */
 export type Supplier<T> = T | (() => T);
 
 export const Supplier = {

src/core/index.ts

@@ -1,4 +1,4 @@
-/** THIS FILE IS MANUALLY MAINAINED: see .fernignore */
+/** THIS FILE IS MANUALLY MAINTAINED: see .fernignore */
 export * from "./streaming-fetcher";
 export * from "./fetcher";
 export * from "./runtime";

src/core/websocket/events.ts

@@ -1,4 +1,4 @@
-/** THIS FILE IS MANUALLY MAINAINED: see .fernignore */
+/** THIS FILE IS MANUALLY MAINTAINED: see .fernignore */
 export class Event {
     public target: any;
     public type: string;

src/core/websocket/index.ts

@@ -1,2 +1,2 @@
-/** THIS FILE IS MANUALLY MAINAINED: see .fernignore */
+/** THIS FILE IS MANUALLY MAINTAINED: see .fernignore */
 export * from "./ws";

src/core/websocket/ws.ts

@@ -1,4 +1,4 @@
-/** THIS FILE IS MANUALLY MAINAINED: see .fernignore */
+/** THIS FILE IS MANUALLY MAINTAINED: see .fernignore */
 import { RUNTIME } from "../runtime";
 import * as Events from "./events";
 import { WebSocket as NodeWebSocket } from "ws";
@@ -33,6 +33,7 @@ export type Options = {
     maxEnqueuedMessages?: number;
     startClosed?: boolean;
     debug?: boolean;
+    shouldAttemptReconnectHook?: (event: CloseEvent) => boolean;
 };
 
 const DEFAULT = {
@@ -74,6 +75,7 @@ export class ReconnectingWebSocket {
     private _binaryType: BinaryType = "blob";
     private _closeCalled = false;
     private _messageQueue: Message[] = [];
+    private _queryParamOverrides = new Map<string, string>();
 
     private readonly _url: UrlProvider;
     private readonly _protocols?: string | string[];
@@ -298,6 +300,36 @@ export class ReconnectingWebSocket {
         }
     }
 
+    /**
+     * Sets or removes a query parameter to be automatically applied to the URL just before connection attempts.
+     * Setting a parameter here overrides any parameter with the same key that might be present in the original URL provider.
+     *
+     * @param key The query parameter key.
+     * @param value The value to set. If null, the override for this key is removed.
+     */
+    public setQueryParamOverride(key: string, value: string | null): void {
+        if (value === null) {
+            if (this._queryParamOverrides.delete(key)) {
+                this._debug(`Removed query parameter override for "${key}"`);
+            }
+        } else {
+            this._debug(`Setting query parameter override: ${key}=${value}`);
+            this._queryParamOverrides.set(key, value);
+        }
+    }
+
+    private _applyQueryParamOverrides(url: string) {
+        let finalUrlString = url;
+        if (this._queryParamOverrides.size > 0) {
+            const finalUrl = new URL(url);
+            for (const [key, value] of this._queryParamOverrides.entries()) {
+                finalUrl.searchParams.set(key, value);
+            }
+            finalUrlString = finalUrl.toString();
+        }
+        return finalUrlString;
+    }
+
     private _debug(...args: any[]) {
         if (this._options.debug) {
             // not using spread because compiled version uses Symbols
@@ -372,6 +404,7 @@ export class ReconnectingWebSocket {
         }
         this._wait()
             .then(() => this._getNextUrl(this._url))
+            .then((url) => this._applyQueryParamOverrides(url))
             .then((url) => {
                 // close could be called before creating the ws
                 if (this._closeCalled) {
@@ -469,10 +502,25 @@ export class ReconnectingWebSocket {
         this._debug("close event");
         this._clearTimeouts();
 
+        let finalShouldReconnect = this._shouldReconnect;
+
         if (event.code === 1000) {
-            this._shouldReconnect = false;
+            finalShouldReconnect = false;
         }
 
+        if (finalShouldReconnect && this._options.shouldAttemptReconnectHook) {
+            try {
+                if (!this._options.shouldAttemptReconnectHook(event)) {
+                    finalShouldReconnect = false;
+                }
+            } catch (e) {
+                console.error("Error executing shouldAttemptReconnectHook:", e);
+                finalShouldReconnect = false;
+            }
+        }
+
+        this._shouldReconnect = finalShouldReconnect;
+
         if (this._shouldReconnect) {
             this._connect();
         }

src/errors/HumeError.ts

@@ -1,4 +1,4 @@
-/** THIS FILE IS MANUALLY MAINAINED: see .fernignore */
+/** THIS FILE IS MANUALLY MAINTAINED: see .fernignore */
 export class HumeError extends Error {
     readonly statusCode?: number;
     readonly body?: unknown;

src/index.ts

@@ -1,4 +1,4 @@
-/** THIS FILE IS MANUALLY MAINAINED: see .fernignore */
+/** THIS FILE IS MANUALLY MAINTAINED: see .fernignore */
 export * as Hume from "./api";
 export * from "./wrapper";
 export { HumeEnvironment } from "./environments";

tests/empathicVoice/chat.test.ts

@@ -1,4 +1,4 @@
-/** THIS FILE IS MANUALLY MAINAINED: see .fernignore */
+/** THIS FILE IS MANUALLY MAINTAINED: see .fernignore */
 import { HumeClient } from "../../src/";
 
 describe("Empathic Voice Interface", () => {

tests/expressionMeasurement/batch.test.ts

@@ -1,4 +1,4 @@
-/** THIS FILE IS MANUALLY MAINAINED: see .fernignore */
+/** THIS FILE IS MANUALLY MAINTAINED: see .fernignore */
 import { HumeClient } from "../../src/";
 
 describe("Streaming Expression Measurement", () => {