rts-sim-testing-service/third_party/semi_physical_train/semi_physical_train.go

package semi_physical_train

import (
	"fmt"
	"sync"

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

// 半实物仿真列车通信接口
type SemiPhysicalTrain interface {
	tpapi.ThirdPartyApiService
	// 启动半实物仿真消息处理
	Start(manager SemiPhysicalMessageManager)
	// 停止半实物仿真消息处理
	Stop()
	// 发送列车控制消息
	SendTrainControlMessage(info *message.DynamicsTrainInfo)
}

type SemiPhysicalMessageManager interface {
	// 处理半实物仿真列车控制消息
	HandleSemiPhysicalTrainControlMsg(b []byte)
	// 获取半实物启动参数
	GetSemiPhysicalRunConfig() *config.VobcConfig
}

type semiPhysicalTrainImpl struct {
	tpapi.ThirdPartyApiService
	trainControlUdpServer   udp.UdpServer
	trainSpeedInfoUdpClient udp.UdpClient

	state            tpapi.ThirdPartyApiServiceState
	udpDelayRecorder *tpapi.UdpDelayRecorder

	manager   SemiPhysicalMessageManager
	runConfig *config.VobcConfig
}

func (s *semiPhysicalTrainImpl) updateState(state tpapi.ThirdPartyApiServiceState) {
	s.state = state
}

func (s *semiPhysicalTrainImpl) State() tpapi.ThirdPartyApiServiceState {
	return s.state
}

func (s *semiPhysicalTrainImpl) Name() string {
	return Name
}

func (s *semiPhysicalTrainImpl) handleTrainControlMsg(b []byte) {
	s.udpDelayRecorder.RecordInterval()
	// slog.Debug(fmt.Sprintf("半实物列车控制消息近期消息间隔: %v", s.udpDelayRecorder.GetIntervals()))
	handler := s.manager
	if handler != nil {
		handler.HandleSemiPhysicalTrainControlMsg(b)
	}
}

func (s *semiPhysicalTrainImpl) Start(manager SemiPhysicalMessageManager) {
	if manager == nil {
		panic("启动半实物消息服务错误: SemiPhysicalMessageManager不能为nil")
	}
	if s.manager != nil {
		panic("启动半实物消息服务错误: 存在正在运行的任务")
	}
	s.runConfig = manager.GetSemiPhysicalRunConfig()
	if s.runConfig == nil || s.runConfig.Ip == "" || !s.runConfig.Open {
		return
	}
	// 初始化客户端、服务端
	s.initSemiPhysical()
	s.manager = manager
	s.updateState(tpapi.ThirdPartyState_Normal)
	s.udpDelayRecorder.Start()
}

func (s *semiPhysicalTrainImpl) Stop() {
	initMutex.Lock()
	defer initMutex.Unlock()
	s.udpDelayRecorder.Stop()
	if s.trainControlUdpServer != nil {
		s.trainControlUdpServer.Close()
	}
	if s.trainSpeedInfoUdpClient != nil {
		s.trainSpeedInfoUdpClient.Close()
	}
	s.manager = nil
	s.updateState(tpapi.ThirdPartyState_Closed)
}

func (s *semiPhysicalTrainImpl) SendTrainControlMessage(info *message.DynamicsTrainInfo) {
	sendMsg := &message.TrainSpeedMsg{}
	sendMsg.DynamicsDecode(info)
	s.trainSpeedInfoUdpClient.Send(sendMsg.Encode())
}

func (s *semiPhysicalTrainImpl) initSemiPhysical() {
	s.trainSpeedInfoUdpClient = udp.NewClient(fmt.Sprintf("%v:%v", s.runConfig.Ip, s.runConfig.RemotePort))
	s.trainControlUdpServer = udp.NewServer(fmt.Sprintf(":%d", s.runConfig.LocalPort), s.handleTrainControlMsg)
	s.trainControlUdpServer.Listen()
}

var _default *semiPhysicalTrainImpl
var initMutex sync.Mutex

const Name = "半实物仿真列车"
const Interval int = 20

func Default() SemiPhysicalTrain {
	initMutex.Lock()
	defer initMutex.Unlock()
	if _default == nil {
		_default = &semiPhysicalTrainImpl{
			udpDelayRecorder: tpapi.NewUdpDelayRecorder(Interval, func(err error) {
				if err != nil {
					_default.updateState(tpapi.ThirdPartyState_Broken)
				} else {
					_default.updateState(tpapi.ThirdPartyState_Normal)
				}
			}),
		}
	}
	return _default
}