rts-sim-testing-service/dynamics/udp.go

package dynamics

import (
	"container/list"
	"encoding/binary"
	"fmt"
	"github.com/panjf2000/gnet/v2"
	"go.uber.org/zap"
	"joylink.club/bj-rtsts-server/config"
	"net"
	"sync"
	"time"
)

func init() {
	runSendTurnoutStateTask()
}

var (
	m sync.Map
)

type TurnoutInfoFunc func() []*TurnoutInfo

func runSendTurnoutStateTask() {
	go func() {
		tick := time.Tick(50 * time.Millisecond)
		for range tick {
			m.Range(func(key, value any) bool {
				slice := value.(TurnoutInfoFunc)()
				for _, turnoutInfo := range slice {
					err := SendTurnoutInfo(turnoutInfo)
					if err != nil {
						zap.S().Error(err)
					}
				}
				return true
			})
		}
	}()
}

func AddTurnoutInfoFunc(simId string, tiFunc TurnoutInfoFunc) {
	m.Store(simId, tiFunc)
}

func RemoveTurnoutInfoFunc(simId string) {
	m.Delete(simId)
}

// SendTurnoutInfo 发送道岔信息
func SendTurnoutInfo(info *TurnoutInfo) error {
	defer func() {
		if r := recover(); r != nil {
			zap.S().Error("发送道岔信息失败", r)
		}
	}()
	addr := fmt.Sprintf("%v:%v", config.Config.Dynamics.Ip, config.Config.Dynamics.UdpRemotePort)
	remoteAddr, _ := net.ResolveUDPAddr("udp", addr)
	conn, err := net.DialUDP("udp", nil, remoteAddr)
	if err != nil {
		zap.S().Error("UDP通信失败", err)
		return err
	}
	defer func(conn *net.UDPConn) {
		err := conn.Close()
		if err != nil {
			zap.S().Error(err)
		}
	}(conn)
	var data []byte
	data = binary.BigEndian.AppendUint16(data, info.Code)
	var b byte
	if info.NPosition {
		b |= 1 << 7
	}
	if info.RPosition {
		b |= 1 << 6
	}
	data = append(data, b)
	_, err = conn.Write(data)

	return err
}

type udpServer struct {
	gnet.BuiltinEventEngine

	eng       gnet.Engine
	addr      string
	multicore bool

	eventHandlers []gnet.EventHandler
}

func (server *udpServer) OnBoot(eng gnet.Engine) gnet.Action {
	server.eng = eng
	println("udp server with multi-core=%t is listening on %s\n", server.multicore, server.addr)
	zap.S().Infof("udp server with multi-core=%t is listening on %s\n", server.multicore, server.addr)
	return gnet.None
}

// OnTraffic 接收到数据后的解析
func (server *udpServer) OnTraffic(c gnet.Conn) gnet.Action {
	defer func() {
		if r := recover(); r != nil {
			zap.S().Error("udp服务数据解析异常", r)
		}
	}()
	buf, _ := c.Next(-1)
	trainInfo := TrainInfo{}
	trainInfo.LifeSignal = binary.BigEndian.Uint16(buf[0:2])
	trainInfo.Number = buf[2]
	trainInfo.Len = binary.BigEndian.Uint16(buf[3:5])
	trainInfo.Link = buf[5]
	trainInfo.LinkOffset = binary.BigEndian.Uint32(buf[6:10])
	trainInfo.Slope = buf[10]
	b := buf[11]
	trainInfo.UpSlope = (b & (1 << 7)) != 0
	trainInfo.Up = (b & (1 << 6)) != 0
	trainInfo.TotalResistance = float32(binary.BigEndian.Uint16(buf[13:15])) / 100
	trainInfo.AirResistance = float32(binary.BigEndian.Uint16(buf[15:17])) / 100
	trainInfo.SlopeResistance = float32(binary.BigEndian.Uint16(buf[17:19])) / 100
	trainInfo.CurveResistance = float32(binary.BigEndian.Uint16(buf[19:21])) / 100
	trainInfo.Speed = float32(binary.BigEndian.Uint16(buf[21:23])) / 10
	trainInfo.HeadSpeed1 = float32(binary.BigEndian.Uint16(buf[23:25])) / 10
	trainInfo.HeadSpeed2 = float32(binary.BigEndian.Uint16(buf[25:27])) / 10
	trainInfo.TailSpeed1 = float32(binary.BigEndian.Uint16(buf[27:29])) / 10
	trainInfo.TailSpeed2 = float32(binary.BigEndian.Uint16(buf[29:31])) / 10
	trainInfo.HeadRadarSpeed = float32(binary.BigEndian.Uint16(buf[31:33])) / 10
	trainInfo.TailRadarSpeed = float32(binary.BigEndian.Uint16(buf[33:35])) / 10

	for e := handlerList.Front(); e != nil; e = e.Next() {
		handler := e.Value.(TrainInfoHandler)
		handler(&trainInfo)
	}

	return gnet.None
}

var handlerList list.List

type TrainInfoHandler func(info *TrainInfo)

func RegisterTrainInfoHandler(handler TrainInfoHandler) {
	handlerList.PushBack(handler)
}

func RunUdpServer() {
	server := &udpServer{addr: fmt.Sprintf("udp://:%d", config.Config.Dynamics.UdpLocalPort), multicore: false}
	err := gnet.Run(server, server.addr, gnet.WithMulticore(server.multicore))
	zap.L().Fatal("udp服务启动失败", zap.Error(err))
}