[修改]12号线通信方式改为TCP

rts-sim-testing-message

@@ -1 +1 @@
-Subproject commit c1499de7bb6e41b56f53c982d516da30988fbd44
+Subproject commit 0844608c8674ec2988c06ba61e73e7b6afe7d660

third_party/interlock/beijing12/interlock.go

@@ -4,6 +4,7 @@ package beijing12
 import (
 	"context"
 	"fmt"
+	"joylink.club/bj-rtsts-server/third_party/tcp"
 	"log/slog"
 	"runtime/debug"
 	"sync"
@@ -11,9 +12,10 @@ import (
 
 	"joylink.club/bj-rtsts-server/config"
 	"joylink.club/bj-rtsts-server/third_party/message"
-	"joylink.club/bj-rtsts-server/third_party/udp"
 )
 
+const logTag = "[北京12号线联锁通信]"
+
 // 联锁代理通信接口
 type InterlockMessageManager interface {
 	CollectInterlockRelayInfo(code string) *message.InterlockSendMsgPkg
@@ -26,8 +28,6 @@ type InterlockProxy interface {
 	Start(manager InterlockMessageManager)
 	// 停止联锁消息功能
 	Stop()
-	// 发送联锁采集消息
-	SendCollectMessage(b []byte)
 }
 
 var interlockMap = make(map[string]InterlockProxy)
@@ -43,8 +43,7 @@ func Default(c *config.InterlockConfig) InterlockProxy {
 }
 
 type interlockProxy struct {
-	driveInfoUdpServer   udp.UdpServer
-	sendCollectUdpClient udp.UdpClient
+	tcpClient *tcp.TcpClient
 
 	manager               InterlockMessageManager
 	collectInfoTaskCancel context.CancelFunc
@@ -53,7 +52,7 @@ type interlockProxy struct {
 
 // 驱动信息进行转发
 func (i *interlockProxy) handleDriverInfo(b []byte) {
-	slog.Info(fmt.Sprintf("收到联锁驱动继电器数据：%x", b))
+	slog.Info(fmt.Sprintf("%s收到联锁驱动继电器数据：%x", logTag, b))
 	handler := i.manager
 	if handler != nil {
 		handler.HandleInterlockDriverInfo(i.runConfig.Code, b)
@@ -65,17 +64,17 @@ func (i *interlockProxy) Start(manager InterlockMessageManager) {
 		return
 	}
 	if manager == nil {
-		panic("启动联锁消息服务错误: InterlockMessageManager不能为nil")
+		panic(fmt.Sprintf("%s启动联锁消息服务错误: InterlockMessageManager不能为nil", logTag))
 	}
 	if i.manager != nil {
-		panic("启动联锁消息服务错误: 存在正在运行的任务")
+		panic(fmt.Sprintf("%s启动联锁消息服务错误: 存在正在运行的任务", logTag))
 	}
 	i.manager = manager
 	// 初始化客户端、服务端
 	i.initInterlockProxy()
-	ctx, cancle := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(context.Background())
 	go i.collectInfoStateTask(ctx)
-	i.collectInfoTaskCancel = cancle
+	i.collectInfoTaskCancel = cancel
 }
 
 // 采集电路状态发送间隔,单位ms
@@ -88,7 +87,7 @@ var serialNumber uint8
 func (i *interlockProxy) collectInfoStateTask(ctx context.Context) {
 	defer func() {
 		if err := recover(); err != nil {
-			slog.Error("定时发送道岔状态任务异常", "error", err, "stack", string(debug.Stack()))
+			slog.Error(logTag+"定时发送道岔状态任务异常", "error", err, "stack", string(debug.Stack()))
 			debug.PrintStack()
 		}
 	}()
@@ -102,11 +101,11 @@ func (i *interlockProxy) collectInfoStateTask(ctx context.Context) {
 		if collectInfoState != nil {
 			serialNumber++
 			collectInfoState.SetSerialNumber(serialNumber)
-			err := i.sendCollectUdpClient.SendMsg(collectInfoState)
+			err := i.tcpClient.Send(collectInfoState.Encode())
 			if err != nil {
-				slog.Error("向联锁发送继电器状态失败:", err)
+				slog.Error(fmt.Sprintf("%s向联锁发送继电器状态失败：%s", logTag, err))
 			} else {
-				slog.Info(fmt.Sprintf("向联锁发送继电器数据成功：%x", collectInfoState.Encode()))
+				slog.Info(fmt.Sprintf("%s向联锁发送继电器数据成功：%x", logTag, collectInfoState.Encode()))
 			}
 		}
 		time.Sleep(time.Millisecond * InterlockMessageSendInterval)
@@ -117,11 +116,9 @@ func (i *interlockProxy) Stop() {
 	initMutex.Lock()
 	defer initMutex.Unlock()
 	delete(interlockMap, i.runConfig.Code)
-	if i.sendCollectUdpClient != nil {
-		i.sendCollectUdpClient.Close()
-	}
-	if i.driveInfoUdpServer != nil {
-		i.driveInfoUdpServer.Close()
+	i.collectInfoTaskCancel()
+	if i.tcpClient != nil {
+		i.tcpClient.Close()
 	}
 	if i.collectInfoTaskCancel != nil {
 		i.collectInfoTaskCancel()
@@ -129,12 +126,21 @@ func (i *interlockProxy) Stop() {
 	i.manager = nil
 }
 
-func (i *interlockProxy) SendCollectMessage(b []byte) {
-	i.sendCollectUdpClient.Send(b)
+func (i *interlockProxy) initInterlockProxy() {
+	client, err := tcp.StartTcpClient(fmt.Sprintf("%s:%d", i.runConfig.Ip, i.runConfig.RemotePort), i.handleDriveInfo, func(err error) {
+		slog.Error(fmt.Sprintf("%sTCP客户端读取数据出错，终止通信服务：%s", logTag, err))
+		i.Stop()
+	})
+	if err != nil {
+		panic(fmt.Sprintf("%s启动TCP客户端失败：%s", logTag, err))
+	}
+	i.tcpClient = client
 }
 
-func (i *interlockProxy) initInterlockProxy() {
-	i.sendCollectUdpClient = udp.NewClient(fmt.Sprintf("%v:%v", i.runConfig.Ip, i.runConfig.RemotePort))
-	i.driveInfoUdpServer = udp.NewServer(fmt.Sprintf(":%d", i.runConfig.LocalPort), i.handleDriverInfo)
-	i.driveInfoUdpServer.Listen()
+func (i *interlockProxy) handleDriveInfo(n int, data []byte) {
+	data = data[:n]
+	slog.Info(fmt.Sprintf("%s收到联锁驱动继电器数据：%x", logTag, data))
+	if i.manager != nil {
+		i.manager.HandleInterlockDriverInfo(i.runConfig.Code, data)
+	}
 }

third_party/tcp/tcp_client.go

@@ -43,10 +43,14 @@ func StartTcpClient(rAddr string, handler func(n int, data []byte), readErr func
 					slog.Error(fmt.Sprintf("TCP客户端[rAddr:%s]读取数据异常连接可能断开：", rAddr), opErr)
 					client.conning = false
 					readErr(readDataErr)
-				} else if err == io.EOF {
+				} else if readDataErr == io.EOF {
 					slog.Warn(fmt.Sprintf("TCP客户端[rAddr:%s]断开连接：", rAddr))
 					client.conning = false
-					readErr(err)
+					readErr(readDataErr)
+				} else {
+					slog.Error(fmt.Sprintf("TCP客户端[rAddr:%s]读数据出错：%s", raddr, readDataErr))
+					client.conning = false
+					readErr(readDataErr)
 				}
 				return
 			}