rts-sim-testing-service/third_party/interlock/beijing12/interlock.go

// Package beijing12 北京12号线联锁通信
package beijing12

import (
	"context"
	"fmt"
	"log/slog"
	"runtime/debug"
	"sync"
	"time"

	"joylink.club/bj-rtsts-server/config"
	"joylink.club/bj-rtsts-server/third_party/message"
	"joylink.club/bj-rtsts-server/third_party/udp"
)

// 联锁代理通信接口
type InterlockMessageManager interface {
	CollectInterlockRelayInfo(code string) *message.InterlockSendMsgPkg
	HandleInterlockDriverInfo(code string, b []byte)
}

// 联锁接口
type InterlockProxy interface {
	// 启动联锁消息功能
	Start(manager InterlockMessageManager)
	// 停止联锁消息功能
	Stop()
	// 发送联锁采集消息
	SendCollectMessage(b []byte)
}

var interlockMap = make(map[string]InterlockProxy)
var initMutex sync.Mutex

func Default(c *config.InterlockConfig) InterlockProxy {
	initMutex.Lock()
	defer initMutex.Unlock()
	if interlockMap[c.Code] == nil {
		interlockMap[c.Code] = &interlockProxy{runConfig: c}
	}
	return interlockMap[c.Code]
}

type interlockProxy struct {
	driveInfoUdpServer   udp.UdpServer
	sendCollectUdpClient udp.UdpClient

	manager               InterlockMessageManager
	collectInfoTaskCancel context.CancelFunc
	runConfig             *config.InterlockConfig
}

// 驱动信息进行转发
func (i *interlockProxy) handleDriverInfo(b []byte) {
	handler := i.manager
	if handler != nil {
		handler.HandleInterlockDriverInfo(i.runConfig.Code, b)
	}
}

func (i *interlockProxy) Start(manager InterlockMessageManager) {
	if i.runConfig == nil || i.runConfig.Ip == "" || !i.runConfig.Open {
		return
	}
	if manager == nil {
		panic("启动联锁消息服务错误: InterlockMessageManager不能为nil")
	}
	if i.manager != nil {
		panic("启动联锁消息服务错误: 存在正在运行的任务")
	}
	i.manager = manager
	// 初始化客户端、服务端
	i.initInterlockProxy()
	ctx, cancle := context.WithCancel(context.Background())
	go i.collectInfoStateTask(ctx)
	i.collectInfoTaskCancel = cancle
}

// 采集电路状态发送间隔,单位ms
const InterlockMessageSendInterval = 50

// 序列号
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()))
			debug.PrintStack()
		}
	}()
	for {
		select {
		case <-ctx.Done():
			return
		default:
		}
		collectInfoState := i.manager.CollectInterlockRelayInfo(i.runConfig.Code)
		if collectInfoState != nil {
			serialNumber++
			collectInfoState.SetSerialNumber(serialNumber)
			i.sendCollectUdpClient.SendMsg(collectInfoState)
		}
		time.Sleep(time.Millisecond * InterlockMessageSendInterval)
	}
}

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()
	}
	if i.collectInfoTaskCancel != nil {
		i.collectInfoTaskCancel()
	}
	i.manager = nil
}

func (i *interlockProxy) SendCollectMessage(b []byte) {
	i.sendCollectUdpClient.Send(b)
}

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()
}