jx_ld_server/utils/myTool.go

package utils

import (
	"encoding/json"
	"fmt"
	"io/ioutil"
	"net"
	"os"
	"path/filepath"
	"server/dao"
	"server/logger"
	"strings"
	"sync"
	"time"
)

const dataFilePath = "static/data.json"

var InitRegionData []dao.Region

// LoadData1 从文件中加载 Region 数据。
func LoadData1() {
	var regions []dao.Region

	// 尝试从主数据文件读取数据
	data, err := os.ReadFile(dataFilePath)
	if err != nil {
		if !os.IsNotExist(err) {
			logger.Get().Errorf("读取文件失败: %v", err)
			return
		}
		// 如果主数据文件不存在，则尝试从备份文件读取
	} else {
		// 检查文件是否为空
		if len(data) > 0 {
			if err := json.Unmarshal(data, &regions); err != nil {
				logger.Get().Errorf("解析 JSON 失败 (主文件): %v, 原始数据: %s", err, string(data))
				return
			}
		}
	}

	InitRegionData = regions
}

// LoadData 从文件中加载 Region 数据。
func LoadData() ([]dao.Region, error) {

	return InitRegionData, 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

}

var s sync.Mutex

func SaveRegions(regions []dao.Region) {
	s.Lock()
	// 创建一个映射以便快速查找
	regionMap := make(map[string]dao.Region)
	orderSlice := make([]string, 0, len(InitRegionData)) // 用于保存顺序的切片
	for _, region := range InitRegionData {
		regionMap[region.Name] = region
		orderSlice = append(orderSlice, region.Name)
	}

	// 遍历传入的新数据，进行版本控制检查
	for _, newRegion := range regions {
		existingRegion, exists := regionMap[newRegion.Name]
		if exists {
			if existingRegion.Version != newRegion.Version {
				logger.Get().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
			orderSlice = append(orderSlice, newRegion.Name) // 添加新区域到顺序切片
		}
	}

	// 将 map 按照 orderSlice 的顺序转换回 slice
	updatedRegions := make([]dao.Region, 0, len(regionMap))
	for _, name := range orderSlice {
		updatedRegions = append(updatedRegions, regionMap[name])
	}
	InitRegionData = updatedRegions
	s.Unlock()
}

// SaveData 保存数据到文件
func SaveData(path string, newData []dao.Region) error {
	// 加载当前数据
	currentRegions, err := LoadData()
	if err != nil {
		return err
	}

	// 创建一个映射以便快速查找
	regionMap := make(map[string]dao.Region)
	orderSlice := make([]string, 0, len(currentRegions)) // 用于保存顺序的切片
	for _, region := range currentRegions {
		regionMap[region.Name] = region
		orderSlice = append(orderSlice, region.Name)
	}

	// 遍历传入的新数据，进行版本控制检查
	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
			orderSlice = append(orderSlice, newRegion.Name) // 添加新区域到顺序切片
		}
	}

	// 将 map 按照 orderSlice 的顺序转换回 slice
	updatedRegions := make([]dao.Region, 0, len(regionMap))
	for _, name := range orderSlice {
		updatedRegions = append(updatedRegions, regionMap[name])
	}

	// 序列化数据
	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.Stat(path); err == nil {
		// 文件存在，检查是否可写
		if writable, err := checkWritable(path); err != nil || !writable {
			return fmt.Errorf("目标文件不可写: %v", err)
		}
	}
	if err = os.Remove(path); err != nil {
		if !os.IsNotExist(err) {
			return fmt.Errorf("删除目标文件失败: %w", err)
		}
	}

	// 尝试重命名临时文件到目标位置
	if err := os.Rename(tempFile.Name(), path); err != nil {
		return fmt.Errorf("替换文件失败: %v", err)
	}

	return nil
}

// 辅助函数：检查文件是否可写
func checkWritable(filename string) (bool, error) {
	file, err := os.OpenFile(filename, os.O_WRONLY, 0)
	if err != nil {
		return false, err
	}
	file.Close()
	return true, 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
}