feat: add promise for connection open

Author: tejasbadadare

lazer/sdk/js/examples/index.ts

@@ -6,65 +6,82 @@ import { PythLazerClient } from "../src/index.js";
 // Ignore debug messages
 console.debug = () => {};
 
-const client = new PythLazerClient(
-  ["wss://pyth-lazer.dourolabs.app/v1/stream"],
-  "access_token",
-  3, // Optionally specify number of parallel redundant connections to reduce the chance of dropped messages. The connections will round-robin across the provided URLs. Default is 3.
-  console // Optionally log socket operations (to the console in this case.)
-);
+async function main() {
+  try {
+    const client = await PythLazerClient.create(
+      ["wss://pyth-lazer.dourolabs.app/v1/stream"],
+      "access_token",
+      3, // Optionally specify number of parallel redundant connections to reduce the chance of dropped messages. The connections will round-robin across the provided URLs. Default is 3.
+      console // Optionally log socket operations (to the console in this case.)
+    );
 
-client.addMessageListener((message) => {
-  console.info("got message:", message);
-  switch (message.type) {
-    case "json": {
-      if (message.value.type == "streamUpdated") {
-        console.info(
-          "stream updated for subscription",
-          message.value.subscriptionId,
-          ":",
-          message.value.parsed?.priceFeeds
-        );
+    client.addMessageListener((message) => {
+      console.info("got message:", message);
+      switch (message.type) {
+        case "json": {
+          if (message.value.type == "streamUpdated") {
+            console.info(
+              "stream updated for subscription",
+              message.value.subscriptionId,
+              ":",
+              message.value.parsed?.priceFeeds
+            );
+          }
+          break;
+        }
+        case "binary": {
+          if ("solana" in message.value) {
+            console.info(
+              "solana message:",
+              message.value.solana?.toString("hex")
+            );
+          }
+          if ("evm" in message.value) {
+            console.info("evm message:", message.value.evm?.toString("hex"));
+          }
+          break;
+        }
       }
-      break;
-    }
-    case "binary": {
-      if ("solana" in message.value) {
-        console.info("solana message:", message.value.solana?.toString("hex"));
-      }
-      if ("evm" in message.value) {
-        console.info("evm message:", message.value.evm?.toString("hex"));
-      }
-      break;
-    }
-  }
-});
+    });
 
-// Create and remove one or more subscriptions on the fly
-await client.subscribe({
-  type: "subscribe",
-  subscriptionId: 1,
-  priceFeedIds: [1, 2],
-  properties: ["price"],
-  chains: ["solana"],
-  deliveryFormat: "binary",
-  channel: "fixed_rate@200ms",
-  parsed: false,
-  jsonBinaryEncoding: "base64",
-});
-await client.subscribe({
-  type: "subscribe",
-  subscriptionId: 2,
-  priceFeedIds: [1, 2, 3, 4, 5],
-  properties: ["price"],
-  chains: ["evm"],
-  deliveryFormat: "json",
-  channel: "fixed_rate@200ms",
-  parsed: true,
-  jsonBinaryEncoding: "hex",
-});
+    // Create and remove one or more subscriptions on the fly
+    await client.subscribe({
+      type: "subscribe",
+      subscriptionId: 1,
+      priceFeedIds: [1, 2],
+      properties: ["price"],
+      chains: ["solana"],
+      deliveryFormat: "binary",
+      channel: "fixed_rate@200ms",
+      parsed: false,
+      jsonBinaryEncoding: "base64",
+    });
+    await client.subscribe({
+      type: "subscribe",
+      subscriptionId: 2,
+      priceFeedIds: [1, 2, 3, 4, 5],
+      properties: ["price"],
+      chains: ["evm"],
+      deliveryFormat: "json",
+      channel: "fixed_rate@200ms",
+      parsed: true,
+      jsonBinaryEncoding: "hex",
+    });
+
+    await new Promise((resolve) => setTimeout(resolve, 10_000));
 
-await new Promise((resolve) => setTimeout(resolve, 10_000));
+    await client.unsubscribe(1);
+    await client.unsubscribe(2);
 
-await client.unsubscribe(1);
-await client.unsubscribe(2);
-client.shutdown();
+    await new Promise((resolve) => setTimeout(resolve, 10_000));
+    client.shutdown();
+  } catch (error) {
+    console.error("Error initializing client:", error);
+    process.exit(1);
+  }
+}
+
+main().catch((error) => {
+  console.error("Unhandled error:", error);
+  process.exit(1);
+});

lazer/sdk/js/src/client.ts

@@ -30,7 +30,7 @@ const UINT32_NUM_BYTES = 4;
 const UINT64_NUM_BYTES = 8;
 
 export class PythLazerClient {
-  wsp: WebSocketPool;
+  private constructor(private readonly wsp: WebSocketPool) {}
 
   /**
    * Creates a new PythLazerClient instance.
@@ -39,13 +39,14 @@ export class PythLazerClient {
    * @param numConnections - The number of parallel WebSocket connections to establish (default: 3). A higher number gives a more reliable stream.
    * @param logger - Optional logger to get socket level logs. Compatible with most loggers such as the built-in console and `bunyan`.
    */
-  constructor(
+  static async create(
     urls: string[],
     token: string,
     numConnections = 3,
     logger: Logger = dummyLogger
-  ) {
-    this.wsp = new WebSocketPool(urls, token, numConnections, logger);
+  ): Promise<PythLazerClient> {
+    const wsp = await WebSocketPool.create(urls, token, numConnections, logger);
+    return new PythLazerClient(wsp);
   }
 
   addMessageListener(handler: (event: JsonOrBinaryResponse) => void) {

lazer/sdk/js/src/socket/resilient-web-socket.ts

@@ -1,21 +1,10 @@
 import type { ClientRequestArgs } from "node:http";
-
 import WebSocket, { type ClientOptions, type ErrorEvent } from "isomorphic-ws";
 import type { Logger } from "ts-log";
 
-// Reconnect with expo backoff if we don't get a message or ping for 10 seconds
 const HEARTBEAT_TIMEOUT_DURATION = 10_000;
+const CONNECTION_TIMEOUT = 5000;
 
-/**
- * This class wraps websocket to provide a resilient web socket client.
- *
- * It will reconnect if connection fails with exponential backoff. Also, it will reconnect
- * if it receives no ping request or regular message from server within a while as indication
- * of timeout (assuming the server sends either regularly).
- *
- * This class also logs events if logger is given and by replacing onError method you can handle
- * connection errors yourself (e.g: do not retry and close the connection).
- */
 export class ResilientWebSocket {
   endpoint: string;
   wsClient: undefined | WebSocket;
@@ -24,10 +13,14 @@ export class ResilientWebSocket {
   private wsFailedAttempts: number;
   private heartbeatTimeout: undefined | NodeJS.Timeout;
   private logger: undefined | Logger;
+  private connectionPromise: Promise<void> | undefined;
+  private resolveConnection: (() => void) | undefined;
+  private rejectConnection: ((error: Error) => void) | undefined;
 
   onError: (error: ErrorEvent) => void;
   onMessage: (data: WebSocket.Data) => void;
   onReconnect: () => void;
+
   constructor(
     endpoint: string,
     wsOptions?: ClientOptions | ClientRequestArgs,
@@ -64,23 +57,47 @@ export class ResilientWebSocket {
     }
   }
 
-  startWebSocket(): void {
+  async startWebSocket(): Promise<void> {
     if (this.wsClient !== undefined) {
-      return;
+      // If there's an existing connection attempt, wait for it
+      if (this.connectionPromise) {
+        return this.connectionPromise;
+      }
+      return Promise.resolve();
     }
 
     this.logger?.info(`Creating Web Socket client`);
 
+    // Create a new promise for this connection attempt
+    this.connectionPromise = new Promise((resolve, reject) => {
+      this.resolveConnection = resolve;
+      this.rejectConnection = reject;
+    });
+
+    // Set a connection timeout
+    const timeoutId = setTimeout(() => {
+      if (this.rejectConnection) {
+        this.rejectConnection(
+          new Error(`Connection timeout after ${CONNECTION_TIMEOUT}ms`)
+        );
+      }
+    }, CONNECTION_TIMEOUT);
+
     this.wsClient = new WebSocket(this.endpoint, this.wsOptions);
     this.wsUserClosed = false;
 
     this.wsClient.addEventListener("open", () => {
       this.wsFailedAttempts = 0;
       this.resetHeartbeat();
+      clearTimeout(timeoutId);
+      this.resolveConnection?.();
     });
 
     this.wsClient.addEventListener("error", (event) => {
       this.onError(event);
+      if (this.rejectConnection) {
+        this.rejectConnection(new Error("WebSocket connection failed"));
+      }
     });
 
     this.wsClient.addEventListener("message", (event) => {
@@ -89,24 +106,23 @@ export class ResilientWebSocket {
     });
 
     this.wsClient.addEventListener("close", () => {
+      clearTimeout(timeoutId);
+      if (this.rejectConnection) {
+        this.rejectConnection(new Error("WebSocket closed before connecting"));
+      }
       void this.handleClose();
     });
 
-    // Handle ping events if supported (Node.js only)
     if ("on" in this.wsClient) {
-      // Ping handler is undefined in browser side
       this.wsClient.on("ping", () => {
         this.logger?.info("Ping received");
         this.resetHeartbeat();
       });
     }
+
+    return this.connectionPromise;
   }
 
-  /**
-   * Reset the heartbeat timeout. This is called when we receive any message (ping or regular)
-   * from the server. If we don't receive any message within HEARTBEAT_TIMEOUT_DURATION,
-   * we assume the connection is dead and reconnect.
-   */
   private resetHeartbeat(): void {
     if (this.heartbeatTimeout !== undefined) {
       clearTimeout(this.heartbeatTimeout);
@@ -145,6 +161,10 @@ export class ResilientWebSocket {
     } else {
       this.wsFailedAttempts += 1;
       this.wsClient = undefined;
+      this.connectionPromise = undefined;
+      this.resolveConnection = undefined;
+      this.rejectConnection = undefined;
+
       const waitTime = expoBackoff(this.wsFailedAttempts);
 
       this.logger?.error(
@@ -163,7 +183,7 @@ export class ResilientWebSocket {
       return;
     }
 
-    this.startWebSocket();
+    await this.startWebSocket();
     await this.waitForMaybeReadyWebSocket();
 
     if (this.wsClient === undefined) {
@@ -180,6 +200,9 @@ export class ResilientWebSocket {
     if (this.wsClient !== undefined) {
       const client = this.wsClient;
       this.wsClient = undefined;
+      this.connectionPromise = undefined;
+      this.resolveConnection = undefined;
+      this.rejectConnection = undefined;
       client.close();
     }
     this.wsUserClosed = true;