پخش کینباز زمان واقعی با Node.JS WebSocket و TypeScript بومی

{
  "type": "subscribe",
  "channels": [
    {
      "name": "ticker",
      "product_ids": ["BTC-USD", "ETH-USD", "BNB-USD"]
    }
  ]
}

{
  "type": "ticker",
  "sequence": 123456789,
  "product_id": "BTC-USD",
  "price": "50000.12",
  "open_24h": "49000.00",
  "volume_24h": "1200.3456",
  "low_24h": "48000.00",
  "high_24h": "51000.00",
  "side": "buy",         // "buy" or "sell"
  "time": "2021-12-01T12:34:56.789Z",
  "trade_id": 12345,
  "last_size": "0.01"   
}

   npm init -y
   npm install --save-dev typescript @types/node
   npx tsc --init

// List of Coinbase products to subscribe to.
const products = ["BTC-USD", "ETH-USD", "BNB-USD"];
const WS_URL = "wss://ws-feed.pro.coinbase.com";

let ws: WebSocket | null = null;
let reconnectTimeout: NodeJS.Timeout | null = null;

/**
 * Connect (or reconnect) to Coinbase’s WebSocket feed.
 */
function connect() {
  ws = new WebSocket(WS_URL);

  ws.onopen = () => {
    console.log("Connected to Coinbase WebSocket API");
    // Send the subscription message upon connection.
    const subscribeMessage = {
      type: "subscribe",
      channels: [
        {
          name: "ticker",
          product_ids: products,
        },
      ],
    };
    ws?.send(JSON.stringify(subscribeMessage));
  };

  ws.onmessage = (event: MessageEvent) => {
    try {
      const message = JSON.parse(event.data.toString());
      // Coinbase emits many types of messages (e.g. subscriptions, heartbeats).
      // We are interested in "ticker" messages which provide real‑time pricing updates.
      if (message.type === "ticker") {
        const {
          product_id,
          price,
          open_24h,
          volume_24h,
          low_24h,
          high_24h,
          side,
          time,
          trade_id,
          last_size,
        } = message;
        console.log(
          `Ticker update for ${product_id} at ${time}: Price=${price}, Open=${open_24h}, ` +
            `High=${high_24h}, Low=${low_24h}, Volume=${volume_24h}, Last Size=${last_size}, Side=${side}`
        );
        // Process the ticker data as needed.
      } else {
        // Optionally log non-ticker messages or handle them accordingly.
        // console.log("Received message:", message);
      }
    } catch (err) {
      console.error("Error parsing incoming message:", err);
    }
  };

  ws.onerror = (event: Event) => {
    console.error("WebSocket error occurred:", event);
    cleanupAndReconnect();
  };

  ws.onclose = (event: CloseEvent) => {
    console.warn(
      `WebSocket closed. Code: ${event.code}, Reason: ${event.reason}`
    );
    // For Coinbase, a normal closure should be code 1000.
    if (event.code !== 1000) {
      console.log("Attempting to reconnect in 5 seconds...");
      reconnectTimeout = setTimeout(connect, 5000);
    }
  };
}

/**
 * Clean up the current connection and schedule a reconnection.
 */
function cleanupAndReconnect() {
  if (ws) {
    // Remove event handlers.
    ws.onopen = null;
    ws.onmessage = null;
    ws.onerror = null;
    ws.onclose = null;
    // Close the connection if it is OPEN or CONNECTING.
    if (ws.readyState === WebSocket.OPEN || ws.readyState === WebSocket.CONNECTING) {
      ws.close(1001, "Reconnecting"); // Code 1001: Going Away.
    }
  }
  if (reconnectTimeout) {
    clearTimeout(reconnectTimeout);
  }
  reconnectTimeout = setTimeout(connect, 5000);
}

connect();

/**
 * Gracefully shut down the WebSocket connection on process termination.
 */
function shutdown() {
  console.log("Shutting down gracefully...");
  if (reconnectTimeout) {
    clearTimeout(reconnectTimeout);
  }
  if (ws && ws.readyState === WebSocket.OPEN) {
    // Send an unsubscribe message if needed before closing.
    const unsubscribeMessage = {
      type: "unsubscribe",
      channels: [
        {
          name: "ticker",
          product_ids: products,
        },
      ],
    };
    ws.send(JSON.stringify(unsubscribeMessage));
    ws.close(1000, "Process terminated"); // Code 1000: Normal Closure.
  }
  process.exit(0);
}

process.on("SIGINT", shutdown);
process.on("SIGTERM", shutdown);

// Optional: Dynamic Subscription Management
// If you wish to subscribe or unsubscribe from additional products dynamically, you can send:
// 
// // To subscribe to a new product, for example, LTC-USD:
// ws.send(
//   JSON.stringify({
//     type: "subscribe",
//     channels: [{ name: "ticker", product_ids: ["LTC-USD"] }]
//   })
// );
// 
// // To unsubscribe from a product:
// ws.send(
//   JSON.stringify({
//     type: "unsubscribe",
//     channels: [{ name: "ticker", product_ids: ["LTC-USD"] }]
//   })
// );