WebSocket - Data Stream

import EventEmitter from "eventemitter3";
 
class WebSocketService {
  constructor(wsUrl) {
    this.wsUrl = wsUrl;
    this.socket = null;
    this.transactionSocket = null;
    this.reconnectAttempts = 0;
    this.reconnectDelay = 2500;
    this.reconnectDelayMax = 4500;
    this.randomizationFactor = 0.5;
    this.emitter = new EventEmitter();
    this.subscribedRooms = new Set();
    this.transactions = new Set();
    this.connect();
 
    if (typeof window !== "undefined") {
      window.addEventListener("beforeunload", this.disconnect.bind(this));
    }
  }
 
  async connect() {
    if (this.socket && this.transactionSocket) {
      return;
    }
 
    try {
      this.socket = new WebSocket(this.wsUrl);
      this.transactionSocket = new WebSocket(this.wsUrl);
      this.setupSocketListeners(this.socket, "main");
      this.setupSocketListeners(this.transactionSocket, "transaction");
    } catch (e) {
      console.error("Error connecting to WebSocket:", e);
      this.reconnect();
    }
  }
 
  setupSocketListeners(socket, type) {
    socket.onopen = () => {
      console.log(`Connected to ${type} WebSocket server`);
      this.reconnectAttempts = 0;
      this.resubscribeToRooms();
    };
 
    socket.onclose = () => {
      console.log(`Disconnected from ${type} WebSocket server`);
      if (type === "main") this.socket = null;
      if (type === "transaction") this.transactionSocket = null;
      this.reconnect();
    };
 
    socket.onmessage = (event) => {
      try {
        const message = JSON.parse(event.data);
        if (message.type === "message") {
          if (message.data?.tx && this.transactions.has(message.data.tx)) {
            return;
          } else if (message.data?.tx) {
            this.transactions.add(message.data.tx);
          }
          if (message.room.includes('price:')) {
            this.emitter.emit(`price-by-token:${message.data.token}`, message.data);
          }
          this.emitter.emit(message.room, message.data);
        }
      } catch (error) {
        console.error("Error processing message:", error);
      }
    };
  }
 
  disconnect() {
    if (this.socket) {
      this.socket.close();
      this.socket = null;
    }
    if (this.transactionSocket) {
      this.transactionSocket.close();
      this.transactionSocket = null;
    }
    this.subscribedRooms.clear();
    this.transactions.clear();
  }
 
  reconnect() {
    console.log("Reconnecting to WebSocket server");
    const delay = Math.min(
      this.reconnectDelay * Math.pow(2, this.reconnectAttempts),
      this.reconnectDelayMax
    );
    const jitter = delay * this.randomizationFactor;
    const reconnectDelay = delay + Math.random() * jitter;
 
    setTimeout(() => {
      this.reconnectAttempts++;
      this.connect();
    }, reconnectDelay);
  }
 
  joinRoom(room) {
    this.subscribedRooms.add(room);
    const socket = room.includes("transaction")
      ? this.transactionSocket
      : this.socket;
    if (socket && socket.readyState === WebSocket.OPEN) {
      socket.send(JSON.stringify({ type: "join", room }));
    }
  }
 
  leaveRoom(room) {
    this.subscribedRooms.delete(room);
    const socket = room.includes("transaction")
      ? this.transactionSocket
      : this.socket;
    if (socket && socket.readyState === WebSocket.OPEN) {
      socket.send(JSON.stringify({ type: "leave", room }));
    }
  }
 
  on(room, listener) {
    this.emitter.on(room, listener);
  }
 
  off(room, listener) {
    this.emitter.off(room, listener);
  }
 
  getSocket() {
    return this.socket;
  }
 
  resubscribeToRooms() {
    if (
      this.socket &&
      this.socket.readyState === WebSocket.OPEN &&
      this.transactionSocket &&
      this.transactionSocket.readyState === WebSocket.OPEN
    ) {
      for (const room of this.subscribedRooms) {
        const socket = room.includes("transaction")
          ? this.transactionSocket
          : this.socket;
        socket.send(JSON.stringify({ type: "join", room }));
      }
    }
  }
}
 
export default WebSocketService;

{
  "token": {
    "name": "Token Name",
    "symbol": "DANCE",
    "mint": "AmJaZvdNptvofC4qe3tvuBNgqLm65p1of5pk6JFHpump",
    "uri": "https://cf-ipfs.com/ipfs/QmVrh4ER81fns3S4QU48WiBuhiusc1KsCxsM8mSs1bEGPv",
    "decimals": 6,
    "hasFileMetaData": true,
    "createdOn": "https://pump.fun"
  },
  "pools": [
    {
      "liquidity": {
        "quote": 62,
        "usd": 8907.761583907999
      },
      "price": {
        "quote": 2.9853991922957425e-8,
        "usd": 0.000004289229715768062
      },
      "tokenSupply": 1000000000000000,
      "lpBurn": 100,
      "tokenAddress": "AmJaZvdNptvofC4qe3tvuBNgqLm65p1of5pk6JFHpump",
      "marketCap": {
        "quote": 29.853991922957423,
        "usd": 4289.229715768061
      },
      "decimals": 6,
      "security": {
        "freezeAuthority": null,
        "mintAuthority": null
      },
      "quoteToken": "So11111111111111111111111111111111111111112",
      "market": "pumpfun",
      "deployer": "4Rz5xqikxtZ2s7wE9uQ6n2oLXQi6K65XGoYpKxf24Hqo",
      "openTime": 0,
      "poolId": "GmJaZvdNptvofC4qe3tvuBNgqLm65p1of5pk6JFHpump"
    }
  ],
  "events": {
    "30m": {
      "priceChangePercentage": 0
    },
    "1h": {
      "priceChangePercentage": 0
    },
    "4h": {
      "priceChangePercentage": 0
    },
    "24h": {
      "priceChangePercentage": 0
    }
  },
  "risk": {
    "rugged": false,
    "risks": [
      {
        "name": "No social media",
        "description": "This token has no social media links",
        "level": "warning",
        "score": 2000
      },
      {
        "name": "Pump.fun contracts can be changed at any time",
        "description": "Pump.fun contracts can be changed by Pump.fun at any time",
        "level": "warning",
        "score": 10
      },
      {
        "name": "Bonding curve not complete",
        "description": "No raydium liquidity pool, bonding curve not complete",
        "level": "warning",
        "score": 4000
      }
    ],
    "score": 5
  }
}