lyroc/backend-deno/modules/websocket_manager.ts

/**
 * WebSocket 连接管理器 (Deno 版本)
 * 负责管理 WebSocket 连接和消息广播
 */

export interface WebSocketConnection {
  id: string;
  socket: WebSocket;
  isAlive: boolean;
}

export class ConnectionManager {
  private connections: Map<string, WebSocketConnection> = new Map();
  private heartbeatInterval: ReturnType<typeof setInterval> | null = null;

  constructor() {
    this.startHeartbeat();
  }

  /**
   * 添加新连接
   */
  addConnection(id: string, socket: WebSocket): void {
    const connection: WebSocketConnection = {
      id,
      socket,
      isAlive: true,
    };

    this.connections.set(id, connection);
    console.log(`新连接已添加: ${id}, 当前连接数: ${this.connections.size}`);

    // 设置连接关闭处理
    socket.addEventListener("close", () => {
      this.removeConnection(id);
    });

    // 设置心跳响应处理
    socket.addEventListener("message", (event) => {
      try {
        const data = JSON.parse(event.data);
        if (data.type === "ping") {
          connection.isAlive = true;
          this.sendToConnection(id, { type: "pong" });
        }
      } catch (error) {
        console.warn(`处理消息时出错: ${error}`);
      }
    });
  }

  /**
   * 移除连接
   */
  removeConnection(id: string): void {
    if (this.connections.has(id)) {
      this.connections.delete(id);
      console.log(`连接已移除: ${id}, 当前连接数: ${this.connections.size}`);
    }
  }

  /**
   * 向特定连接发送消息
   */
  sendToConnection(id: string, message: any): boolean {
    const connection = this.connections.get(id);
    if (connection && connection.socket.readyState === WebSocket.OPEN) {
      try {
        connection.socket.send(JSON.stringify(message));
        return true;
      } catch (error) {
        console.error(`发送消息到连接 ${id} 失败: ${error}`);
        this.removeConnection(id);
        return false;
      }
    }
    return false;
  }

  /**
   * 向所有连接广播消息
   */
  broadcast(message: any): void {
    const deadConnections: string[] = [];

    for (const [id, connection] of this.connections) {
      if (connection.socket.readyState === WebSocket.OPEN) {
        try {
          connection.socket.send(JSON.stringify(message));
        } catch (error) {
          console.error(`广播消息到连接 ${id} 失败: ${error}`);
          deadConnections.push(id);
        }
      } else {
        deadConnections.push(id);
      }
    }

    // 清理死连接
    deadConnections.forEach((id) => this.removeConnection(id));
  }

  /**
   * 获取连接数量
   */
  getConnectionCount(): number {
    return this.connections.size;
  }

  /**
   * 获取所有连接 ID
   */
  getConnectionIds(): string[] {
    return Array.from(this.connections.keys());
  }

  /**
   * 启动心跳检测
   */
  private startHeartbeat(): void {
    this.heartbeatInterval = setInterval(() => {
      const deadConnections: string[] = [];

      for (const [id, connection] of this.connections) {
        if (!connection.isAlive) {
          deadConnections.push(id);
        } else {
          connection.isAlive = false;
          // 发送心跳检测
          this.sendToConnection(id, { type: "ping" });
        }
      }

      // 清理死连接
      deadConnections.forEach((id) => this.removeConnection(id));
    }, 30000); // 每30秒检测一次
  }

  /**
   * 停止心跳检测
   */
  stopHeartbeat(): void {
    if (this.heartbeatInterval) {
      clearInterval(this.heartbeatInterval);
      this.heartbeatInterval = null;
    }
  }

  /**
   * 关闭所有连接
   */
  closeAllConnections(): void {
    for (const [id, connection] of this.connections) {
      try {
        connection.socket.close();
      } catch (error) {
        console.error(`关闭连接 ${id} 失败: ${error}`);
      }
    }
    this.connections.clear();
    this.stopHeartbeat();
  }
}