package initialize
import (
"encoding/hex"
"net"
"os"
"runtime"
"runtime/debug"
"server/dao"
"server/logger"
"server/modbus"
"server/model"
"server/service"
"server/utils"
"strconv"
"strings"
"sync"
"syscall"
"time"
)
type ModbusHandler struct {
queue *modbus.MlQueue
}
var _handlerOnce sync.Once
var _handlerSingle *ModbusHandler
func GetHandler() *ModbusHandler {
_handlerOnce.Do(func() {
_handlerSingle = &ModbusHandler{
queue: modbus.NewQueue(10000),
}
})
return _handlerSingle
}
func InitInductanceTCP() {
lis, err := net.Listen("tcp", ":60001")
if err != nil {
logger.Get().Fatalln("Failed to listen on port 60001:", err)
} else {
logger.Get().Println("inductanceTCP启动成功")
model.InductanceTCP = lis
}
go StartInductanceTCP()
}
func StartInductanceTCP() {
logger.Get().Println("进入StartInductanceTCP")
handler := GetHandler()
for {
logger.Get().Println("进入InductanceTCP循环")
conn, err := model.InductanceTCP.Accept()
if err != nil {
logger.Get().Errorf("Accept error: %v", err)
// 根据实际情况决定是否继续循环
continue
}
remoteAddr := conn.RemoteAddr().String()
logger.Get().Printf("lis Accept conn = %s\n", remoteAddr)
deviceId, err := GetDeviceId(conn)
if err != nil {
logger.Get().Errorf("Error getting device ID from %s: %v", remoteAddr, err)
conn.Close()
continue
}
if len(deviceId) != 16 {
continue
}
model.Mutex.Lock()
if existingConn, exists := model.ConnectionMap[deviceId]; exists {
logger.Get().Printf("Connection from %s already exists for deviceId %d, closing new connection\n", remoteAddr, deviceId)
existingConn.Close() // 关闭旧连接
delete(model.ConnectionMap, deviceId) // 清理旧连接
}
model.ConnectionMap[deviceId] = conn
logger.Get().Printf("lis Accept conn = %s, 添加进入 %s \n", remoteAddr, deviceId)
model.Mutex.Unlock()
// 使用新的协程处理新的连接
go handler.ReadAndHandle(conn, deviceId)
go handler.Handler()
}
}
func GetDeviceId(conn net.Conn) (id string, err error) {
// 发送 Modbus RTU 帧
//FE 04 03 EE 00 08 85 B2
data, err := utils.WriteAndReadDevice(modbus.ReadDeviceId, conn, 9, 2)
if err != nil {
return
}
return string(data), err
}
func (o *ModbusHandler) ReadAndHandle(conn net.Conn, deviceId string) {
for {
buffer := make([]byte, 1024)
n, err := conn.Read(buffer)
if err != nil {
if isConnReset(err) {
logger.Get().Error("连接被远程主机强制关闭")
} else if os.IsTimeout(err) {
logger.Get().Error("读取操作超时")
} else {
// 处理其他类型的错误
logger.Get().Errorf("读取错误: %s\n", err)
}
return
}
queueData := model.QueueData{
Id: deviceId,
Value: buffer[:n],
}
ok, cnt := o.queue.Put(&queueData)
if ok {
continue
} else {
logger.Get().Errorf("HandlerData:查询队列失败,队列消息数量:%d", cnt)
runtime.Gosched()
}
}
}
func isConnReset(err error) bool {
if opErr, ok := err.(*net.OpError); ok {
if opErr.Err == syscall.ECONNRESET {
return true // Unix-like 系统上的 ECONNRESET
} else if runtime.GOOS == "windows" {
// Windows 上的 WSAECONNRESET 通常是通过错误消息识别的
if se, ok := opErr.Err.(*os.SyscallError); ok {
if errno, ok := se.Err.(syscall.Errno); ok {
if errno == 10054 { // 10054 对应 WSAECONNRESET
return true
}
}
} else if strings.Contains(opErr.Err.Error(), "WSAECONNRESET") {
// 如果错误消息包含 WSAECONNRESET，也认为是连接被重置
return true
}
}
}
return false
}
func (o *ModbusHandler) Handler() interface{} {
defer func() {
if err := recover(); err != nil {
go GetHandler().Handler()
logger.Get().Errorf("MqttHandler.Handler:发生异常:%s", string(debug.Stack()))
}
}()
for {
msg, ok, quantity := o.queue.Get()
if !ok {
time.Sleep(10 * time.Millisecond)
continue
} else if quantity > 1000 {
logger.Get().Errorf("数据队列累积过多,请注意优化,当前队列条数:%d", quantity)
}
queueData, ok := msg.(*model.QueueData)
if !ok {
logger.Get().Errorln("Type assertion failed: msg is not of type model.QueueDat")
return nil
}
// 信息处理返回
parseData(queueData)
// 对数据进行修改
}
}
func parseData(data *model.QueueData) {
reg, dev, err := utils.GetDataByDeviceId(data.Id)
if err != nil {
logger.Get().Errorln("Error getting register and device:", err)
return
}
toString := hex.EncodeToString(data.Value)
switch toString[0:2] {
case "fe":
switch toString[4:8] { // 开关灯
case "0000", "0001", "0002", "0003", "0004", "0005", "0006", "0007":
logger.Get().Println(dev.Sn + "-----" + toString)
zhi, _ := strconv.Atoi(toString[7:8])
if toString[8:12] == "0000" {
dev.DeviceLoops[zhi].State = 0
} else if toString[8:12] == "ff00" {
dev.DeviceLoops[zhi].State = 1
}
for i, device := range reg.Devices {
if device.Sn == data.Id {
reg.Devices[i] = dev
}
}
regions, err := utils.SaveRegionOnData(reg)
err = service.SaveData(regions)
if err != nil {
logger.Get().Errorln("设备回路状态" + err.Error())
return
}
}
switch toString[2:6] {
case "0101":
for i, device := range reg.Devices {
if device.Sn == data.Id {
if reg.Devices[i].State != 1 {
reg.Devices[i].State = 1
service.Cron{}.RelayOnOffTimeTaskSn(data.Id)
}
reg.Devices[i].OnlineTime = time.Now()
two, err := utils.SixteenTurnsTwo(toString[6:8])
if err != nil {
logger.Get().Errorln("16转2进制" + err.Error())
return
}
for i2, _ := range reg.Devices[i].DeviceLoops {
state := string(two[7-i2])
zhi, _ := strconv.Atoi(state)
reg.Devices[i].DeviceLoops[i2].State = zhi
}
}
}
regions, err := utils.SaveRegionOnData(reg)
err = service.SaveData(regions)
if err != nil {
logger.Get().Errorln("设备回路状态" + err.Error())
return
}
}
case "01":
switch toString[2:6] {
case "0336":
batteryVoltage, _ := strconv.ParseInt(toString[6:10], 16, 64)
batteryCurrent, _ := strconv.ParseInt(toString[10:14], 16, 64)
batteryPlateVoltage, _ := strconv.ParseInt(toString[38:42], 16, 64)
sun := dao.Sun{
DeviceId: dev.Sn,
BatteryVoltage: float64(batteryVoltage) / 100,
BatteryCurrent: int(batteryCurrent),
BatteryPlateVoltage: float64(batteryPlateVoltage) / 100,
}
for i, device := range reg.Devices {
if device.Sn == dev.Sn {
reg.Devices[i].Sun = sun
}
}
regions, err := utils.SaveRegionOnData(reg)
err = service.SaveData(regions)
if err != nil {
logger.Get().Println("太阳能数据保存" + err.Error())
return
}
//电池
//if float64(batteryVoltage)/100 < 5 {
// data1 := modbus.DeviceSwitch(8, 1)
// utils.WriteDevice(data1, model.ConnectionMap[data.Id])
//} else {
// data1 := modbus.DeviceSwitch(8, 0)
// utils.WriteDevice(data1, model.ConnectionMap[data.Id])
//}
}
}
switch toString[0:4] {
case "4c43":
bytes, err := hex.DecodeString(toString[4:])
if err != nil {
logger.Get().Errorln("Error decoding bytes:", err)
return
}
for i, device := range reg.Devices {
if device.Sn == string(bytes) {
reg.Devices[i].State = 1
reg.Devices[i].OnlineTime = time.Now()
}
}
regions, err := utils.SaveRegionOnData(reg)
err = service.SaveData(regions)
if err != nil {
logger.Get().Errorln("心跳" + err.Error())
return
}
}
}