Solana WebSocket：实时区块链数据流

class ConnectionMonitor {
  constructor(ws) {
    this.ws = ws;
    this.pingInterval = null;
    this.lastPong = Date.now();
    this.isConnected = false;
    
    this.setupEventListeners();
    this.startPingMonitoring();
  }
  
  setupEventListeners() {
    this.ws.onopen = () => {
      console.log('Connected');
      this.isConnected = true;
      this.lastPong = Date.now();
    };
    
    this.ws.onclose = (event) => {
      console.log('Disconnected:', event.code, event.reason);
      this.isConnected = false;
      this.stopPingMonitoring();
    };
    
    this.ws.onerror = (error) => {
      console.error('WebSocket error:', error);
      this.isConnected = false;
    };
    
    // Listen for pong responses (server acknowledgment)
    this.ws.onmessage = (event) => {
      const data = JSON.parse(event.data);
      if (data.method === 'pong') {
        this.lastPong = Date.now();
      }
      // Handle other messages...
    };
  }
  
  startPingMonitoring() {
    this.pingInterval = setInterval(() => {
      if (this.isConnected) {
        // Send ping to check connection health
        this.ws.send(JSON.stringify({
          jsonrpc: '2.0',
          method: 'ping',
          id: Date.now()
        }));
        
        // Check if we received a pong recently
        const timeSinceLastPong = Date.now() - this.lastPong;
        if (timeSinceLastPong > 30000) { // 30 seconds timeout
          console.warn('No pong received, connection may be stale');
          this.ws.close();
        }
      }
    }, 10000); // Ping every 10 seconds
  }
  
  stopPingMonitoring() {
    if (this.pingInterval) {
      clearInterval(this.pingInterval);
      this.pingInterval = null;
    }
  }
}

class ExponentialBackoffReconnector {
  constructor(url, maxRetries = 10) {
    this.url = url;
    this.maxRetries = maxRetries;
    this.retryCount = 0;
    this.baseDelay = 1000; // Start with 1 second
    this.maxDelay = 30000; // Cap at 30 seconds
    this.ws = null;
    this.subscriptions = new Map();
    this.isReconnecting = false;
  }
  
  connect() {
    if (this.isReconnecting) return;
    
    try {
      this.ws = new WebSocket(this.url);
      this.setupEventHandlers();
    } catch (error) {
      console.error('Failed to create WebSocket:', error);
      this.scheduleReconnect();
    }
  }
  
  setupEventHandlers() {
    this.ws.onopen = () => {
      console.log('Connected successfully');
      this.retryCount = 0; // Reset retry count on successful connection
      this.isReconnecting = false;
      this.resubscribeAll(); // Restore subscriptions
    };
    
    this.ws.onclose = (event) => {
      console.log('Connection closed:', event.code);
      if (!this.isReconnecting) {
        this.scheduleReconnect();
      }
    };
    
    this.ws.onerror = (error) => {
      console.error('WebSocket error:', error);
    };
  }
  
  scheduleReconnect() {
    if (this.retryCount >= this.maxRetries) {
      console.error('Max retry attempts reached. Giving up.');
      return;
    }
    
    this.isReconnecting = true;
    this.retryCount++;
    
    // Calculate delay with exponential backoff + jitter
    const delay = Math.min(
      this.baseDelay * Math.pow(2, this.retryCount - 1),
      this.maxDelay
    );
    
    // Add jitter to prevent thundering herd
    const jitteredDelay = delay + (Math.random() * 1000);
    
    console.log(`Reconnecting in ${jitteredDelay}ms (attempt ${this.retryCount}/${this.maxRetries})`);
    
    setTimeout(() => {
      this.connect();
    }, jitteredDelay);
  }
  
  subscribe(method, params, callback) {
    const id = this.generateId();
    this.subscriptions.set(id, { method, params, callback });
    
    if (this.ws && this.ws.readyState === WebSocket.OPEN) {
      this.sendSubscription(id, method, params);
    }
    
    return id;
  }
  
  resubscribeAll() {
    console.log(`Restoring ${this.subscriptions.size} subscriptions`);
    for (const [id, sub] of this.subscriptions) {
      this.sendSubscription(id, sub.method, sub.params);
    }
  }
  
  sendSubscription(id, method, params) {
    this.ws.send(JSON.stringify({
      jsonrpc: '2.0',
      id: id,
      method: method,
      params: params
    }));
  }
  
  generateId() {
    return Date.now() + Math.random();
  }
}

// Test disconnection scenarios
class NetworkSimulator {
  constructor(wsManager) {
    this.wsManager = wsManager;
  }
  
  // Simulate network outage
  simulateNetworkOutage(duration = 5000) {
    console.log('Simulating network outage...');
    
    // Force close the connection
    if (this.wsManager.ws) {
      this.wsManager.ws.close(1006, 'Network outage simulation');
    }
    
    // Block reconnection temporarily
    const originalConnect = this.wsManager.connect.bind(this.wsManager);
    this.wsManager.connect = () => {
      console.log('Connection blocked during outage simulation');
    };
    
    // Restore connection after duration
    setTimeout(() => {
      console.log('Network restored');
      this.wsManager.connect = originalConnect;
      this.wsManager.connect();
    }, duration);
  }
  
  // Simulate intermittent connectivity
  simulateIntermittentConnectivity() {
    setInterval(() => {
      if (Math.random() < 0.1) { // 10% chance every 10 seconds
        console.log('Simulating connection drop...');
        this.wsManager.ws?.close(1006, 'Intermittent connectivity');
      }
    }, 10000);
  }
}

// Usage
const simulator = new NetworkSimulator(wsManager);
simulator.simulateNetworkOutage(10000); // 10 second outage