package utils import ( "encoding/json" "fmt" "io/ioutil" "net" "os" "path/filepath" "server/dao" "server/logger" "strings" "time" ) const dataFilePath = "static/data.json" // LoadData 从文件中加载 Region 数据。 func LoadData() ([]dao.Region, error) { var regions []dao.Region // 尝试从主数据文件读取数据 data, err := os.ReadFile(dataFilePath) if err != nil { if !os.IsNotExist(err) { return nil, fmt.Errorf("读取文件失败: %v", err) } // 如果主数据文件不存在,则尝试从备份文件读取 } else { // 检查文件是否为空 if len(data) > 0 { if err := json.Unmarshal(data, ®ions); err != nil { return nil, fmt.Errorf("解析 JSON 失败 (主文件): %v, 原始数据: %s", err, string(data)) } } } return regions, nil } func GetOnlineDevices() (devices []dao.Device, err error) { regions, err := LoadData() if err != nil { return nil, err } for _, region := range regions { for _, device := range region.Devices { if device.State == 0 { continue } devices = append(devices, device) } } return devices, err } func GetAllDevices() (devices []dao.Device, err error) { regions, err := LoadData() if err != nil { return nil, err } for _, region := range regions { for _, device := range region.Devices { devices = append(devices, device) } } return devices, err } // GetDataByDeviceId 按设备id获取信息 地区+设备 func GetDataByDeviceId(deviceId string) (reg dao.Region, dev dao.Device, err error) { regions, err := LoadData() if err != nil { return reg, dev, err } for _, region := range regions { for _, device := range region.Devices { if device.Sn == deviceId { return region, device, nil } } } return reg, dev, fmt.Errorf("设备%s未找到", deviceId) } func SaveRegionOnData(data dao.Region) ([]dao.Region, error) { regions, err := LoadData() if err != nil { return regions, err } for i, region := range regions { if region.Name == data.Name { regions[i] = data break } } return regions, nil } // SaveData 保存数据到文件 func SaveData(path string, newData []dao.Region) error { // 加载当前数据 currentRegions, err := LoadData() if err != nil { return err } // 创建一个映射以便快速查找 regionMap := make(map[string]dao.Region) for _, region := range currentRegions { regionMap[region.Name] = region } // 遍历传入的新数据,进行版本控制检查 for _, newRegion := range newData { existingRegion, exists := regionMap[newRegion.Name] if exists { if existingRegion.Version != newRegion.Version { return fmt.Errorf("版本冲突: 区域 '%s' 的当前版本=%d, 提交版本=%d", newRegion.Name, existingRegion.Version, newRegion.Version) } // 更新版本号 newRegion.Version++ regionMap[newRegion.Name] = newRegion } else { // 如果区域不存在,则直接添加(假设这是新增的区域) newRegion.Version = 1 // 新增区域的初始版本号设为1 regionMap[newRegion.Name] = newRegion } } // 将 map 转换回 slice updatedRegions := make([]dao.Region, 0, len(regionMap)) for _, region := range regionMap { updatedRegions = append(updatedRegions, region) } // 序列化数据 data, err := json.MarshalIndent(updatedRegions, "", " ") if err != nil { return fmt.Errorf("JSON 序列化失败: %v", err) } // 创建临时文件用于原子写入 tempFile, err := ioutil.TempFile(filepath.Dir(path), "tmp-") if err != nil { return fmt.Errorf("创建临时文件失败: %v", err) } defer os.Remove(tempFile.Name()) // 确保即使出错也能清理临时文件 // 写入临时文件 if _, err := tempFile.Write(data); err != nil { tempFile.Close() return fmt.Errorf("写入临时文件失败: %v", err) } // 确保所有数据都写入磁盘 if err := tempFile.Sync(); err != nil { tempFile.Close() return fmt.Errorf("同步临时文件失败: %v", err) } // 关闭临时文件 if err := tempFile.Close(); err != nil { return fmt.Errorf("关闭临时文件失败: %v", err) } // 使用原子操作替换原始文件 if err := os.Rename(tempFile.Name(), path); err != nil { return fmt.Errorf("替换文件失败: %v", err) } return nil } const ( maxRetries = 3 // 最大重试次数 readTimeout = 5 * time.Second // 读取超时时间 writeTimeout = 5 * time.Second // 写入超时时间 reconnectWait = 2 * time.Second // 重连等待时间 ) // checkConnection 检查连接是否仍然有效 func checkConnection(conn net.Conn) error { probe := []byte{} _, err := conn.Write(probe) return err } // checkAndReconnect 检查是否需要重连,并在必要时执行重连 func checkAndReconnect(conn net.Conn) (net.Conn, error) { remoteAddr := conn.RemoteAddr().String() // 关闭旧连接 if err := conn.Close(); err != nil { logger.Get().Errorf("Failed to close connection: %v", err) return nil, err } // 解析原始连接的远程地址和网络接口 addr, networkInterface, err := parseRemoteAddr(remoteAddr) if err != nil { logger.Get().Errorf("Failed to parse remote address: %v", err) return nil, err } // 尝试重新建立连接 newConn, err := net.Dial("tcp", fmt.Sprintf("%s%%%s", addr, networkInterface)) if err != nil { logger.Get().Errorf("Reconnect failed: %v", err) return nil, err } logger.Get().Warnf("Successfully reconnected to %s", remoteAddr) return newConn, nil } // isConnectionClosedError 检查错误是否表明连接被对端强制关闭或已关闭 func isConnectionClosedError(err error) bool { if ne, ok := err.(*net.OpError); ok { return strings.Contains(ne.Err.Error(), "forcibly closed") || strings.Contains(ne.Err.Error(), "broken pipe") || strings.Contains(ne.Err.Error(), "connection reset") || strings.Contains(ne.Err.Error(), "use of closed network connection") } return false } // setDeadlineWithRetry 使用重试机制设置读写截止时间 func setDeadlineWithRetry(conn net.Conn, timeout time.Duration, operation string) error { for attempts := 0; attempts < maxRetries; attempts++ { var err error switch operation { case "read": err = conn.SetReadDeadline(time.Now().Add(timeout)) case "write": err = conn.SetWriteDeadline(time.Now().Add(timeout)) default: return fmt.Errorf("invalid operation: %s", operation) } if err == nil { return nil } if isConnectionClosedError(err) { logger.Get().Warnf("Set %s deadline failed due to closed connection, retrying (%d/%d)", operation, attempts+1, maxRetries) var newConn net.Conn var reconnErr error newConn, reconnErr = checkAndReconnect(conn) if reconnErr != nil { time.Sleep(reconnectWait) // 等待一段时间后重试 continue // 继续下一次重试 } conn = newConn continue // 重试设置截止时间 } logger.Get().Errorf("Set %s deadline failed: %v", operation, err) return fmt.Errorf("set %s deadline failed after %d retries: %v", operation, attempts+1, err) } return fmt.Errorf("failed to set %s deadline after %d retries", operation, maxRetries) } // readWithRetry 使用重试机制进行读取操作 func readWithRetry(buffer []byte, conn net.Conn) (int, error) { for attempts := 0; attempts < maxRetries; attempts++ { if err := setDeadlineWithRetry(conn, readTimeout, "read"); err != nil { return 0, err } // 检查连接是否仍然有效 if err := checkConnection(conn); err != nil { if isConnectionClosedError(err) { logger.Get().Warnf("Connection check failed due to closed connection, retrying (%d/%d)", attempts+1, maxRetries) var newConn net.Conn var reconnErr error newConn, reconnErr = checkAndReconnect(conn) if reconnErr != nil { time.Sleep(reconnectWait) // 等待一段时间后重试 continue // 继续下一次重试 } conn = newConn continue // 重试读取 } logger.Get().Errorf("Connection check failed: %v", err) return 0, fmt.Errorf("connection check failed after %d retries: %v", attempts+1, err) } n, err := conn.Read(buffer) if err == nil { return n, nil } if isConnectionClosedError(err) { logger.Get().Warnf("Read failed due to closed connection, retrying (%d/%d)", attempts+1, maxRetries) var newConn net.Conn var reconnErr error newConn, reconnErr = checkAndReconnect(conn) if reconnErr != nil { time.Sleep(reconnectWait) // 等待一段时间后重试 continue // 继续下一次重试 } conn = newConn continue // 重试读取 } logger.Get().Errorf("Read failed: %v", err) return 0, fmt.Errorf("read failed after %d retries: %v", attempts+1, err) } return 0, fmt.Errorf("failed to read after %d retries", maxRetries) } // writeWithRetry 使用重试机制进行写入操作 func writeWithRetry(frame []byte, conn net.Conn) error { for attempts := 0; attempts < maxRetries; attempts++ { if err := setDeadlineWithRetry(conn, writeTimeout, "write"); err != nil { return err } // 检查连接是否仍然有效 if err := checkConnection(conn); err != nil { if isConnectionClosedError(err) { logger.Get().Warnf("Connection check failed due to closed connection, retrying (%d/%d)", attempts+1, maxRetries) var newConn net.Conn var reconnErr error newConn, reconnErr = checkAndReconnect(conn) if reconnErr != nil { time.Sleep(reconnectWait) // 等待一段时间后重试 continue // 继续下一次重试 } conn = newConn continue // 重试写入 } logger.Get().Errorf("Connection check failed: %v", err) return fmt.Errorf("connection check failed after %d retries: %v", attempts+1, err) } _, err := conn.Write(frame) if err == nil { return nil } if isConnectionClosedError(err) { logger.Get().Warnf("Write failed due to closed connection, retrying (%d/%d)", attempts+1, maxRetries) var newConn net.Conn var reconnErr error newConn, reconnErr = checkAndReconnect(conn) if reconnErr != nil { time.Sleep(reconnectWait) // 等待一段时间后重试 continue // 继续下一次重试 } conn = newConn continue // 重试写入 } logger.Get().Errorf("Write failed: %v", err) return fmt.Errorf("write failed after %d retries: %v", attempts+1, err) } return fmt.Errorf("failed to write after %d retries", maxRetries) } // ReadDevice 从设备读取数据 func ReadDevice(buffer []byte, conn net.Conn) (int, error) { return readWithRetry(buffer, conn) } // WriteDevice1 向设备写入数据 func WriteDevice1(frame []byte, conn net.Conn) error { return writeWithRetry(frame, conn) } func WriteDevice(frame []byte, conn net.Conn) error { for attempts := 0; attempts < maxRetries; attempts++ { if err := conn.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil { logger.Get().Errorf("Set write deadline failed: %v", err) return err } _, err := conn.Write(frame) if err == nil { return nil } // 检查是否是对端强制关闭连接的错误 if ne, ok := err.(*net.OpError); ok && (strings.Contains(ne.Err.Error(), "forcibly closed") || strings.Contains(ne.Err.Error(), "broken pipe") || strings.Contains(ne.Err.Error(), "connection reset")) { logger.Get().Warnf("Connection forcibly closed by peer, retrying (%d/%d)", attempts+1, maxRetries) // 关闭旧连接 if err := conn.Close(); err != nil { logger.Get().Errorf("Failed to close connection: %v", err) } // 尝试重新建立连接 var newConn net.Conn remoteAddr := conn.RemoteAddr().String() // 解析原始连接的远程地址和网络接口 addr, networkInterface, err := parseRemoteAddr(remoteAddr) if err != nil { logger.Get().Errorf("Failed to parse remote address: %v", err) continue // 继续下一次重试 } newConn, err = net.Dial("tcp", fmt.Sprintf("%s%%%s", addr, networkInterface)) if err != nil { logger.Get().Errorf("Reconnect failed: %v", err) time.Sleep(reconnectWait) // 等待一段时间后重试 continue // 继续下一次重试 } conn = newConn continue // 重试写入 } // 如果不是对端强制关闭连接的错误,则直接返回错误 logger.Get().Errorf("Write failed: %v", err) return fmt.Errorf("write failed after %d retries: %v", attempts+1, err) } return fmt.Errorf("failed to write after %d retries", maxRetries) } // 解析远程地址,提取 IP 地址和网络接口名称 func parseRemoteAddr(addr string) (string, string, error) { parts := strings.Split(addr, "%") if len(parts) != 2 { return "", "", fmt.Errorf("invalid remote address format: %s", addr) } return parts[0], parts[1], nil } func WriteAndReadDevice(frame []byte, conn net.Conn, former, after int) (data []byte, err error) { // 发送 Modbus RTU 帧 n, err := conn.Write(frame) if err != nil { logger.Get().Errorln("Error writing to connection:", err) return } // 等待一段时间以接收响应 time.Sleep(1000 * time.Millisecond) // 读取响应 buffer := make([]byte, 1024) n, err = conn.Read(buffer) if err != nil { logger.Get().Errorln("Error reading from connection:", err) return } // 检查读取的字节数是否足够 if n < former+after { err = fmt.Errorf("not enough bytes read to satisfy the slice range") return } // 返回子切片 return buffer[former : n-after], err }