Merge branch 'master' of https://git.code.tencent.com/beijing-rtss-test/bj-rtsts-server-go

api/simulation.go

@@ -1,13 +1,14 @@
 package api
 
 import (
-	"fmt"
 	"net/http"
 
 	jwt "github.com/appleboy/gin-jwt/v2"
 	"github.com/gin-gonic/gin"
 	"go.uber.org/zap"
+	"joylink.club/bj-rtsts-server/ats/verify/simulation"
 	"joylink.club/bj-rtsts-server/dto"
+	apiproto "joylink.club/bj-rtsts-server/grpcproto"
 )
 
 func InitSimulationRouter(api *gin.RouterGroup, authMiddleware *jwt.GinJWTMiddleware) {
@@ -19,6 +20,8 @@ func InitSimulationRouter(api *gin.RouterGroup, authMiddleware *jwt.GinJWTMiddle
 	authed.POST("/check/data", checkSimMapData)
 	authed.POST("/train/remove", removeTrain)
 	authed.POST("/switch/operation", switchOperation)
+
+	apiproto.RegisterMsgServer(&apiproto.SimulationServer{})
 }
 
 // 创建ATS测试仿真
@@ -45,7 +48,8 @@ func create(c *gin.Context) {
 	zap.S().Debug("创建仿真请求：", req)
 	rsp := dto.SimulationCreateRspDto{}
 	rsp.MapId = req.MapId
-	rsp.SimulationId = fmt.Sprintf("sim-%d", req.MapId)
+	//rsp.SimulationId = fmt.Sprintf("sim-%d", req.MapId)
+	rsp.SimulationId = simulation.CreateSimulation(1, nil)
 	c.JSON(http.StatusOK, &rsp)
 }
 

ats/verify/simulation/simulation_manage.go

@@ -0,0 +1,48 @@
+package simulation
+
+import (
+	"strconv"
+
+	"joylink.club/bj-rtsts-server/ats/verify/protos/graphicData"
+	"joylink.club/bj-rtsts-server/ats/verify/protos/state"
+	"joylink.club/bj-rtsts-server/ats/verify/simulation/wayside"
+)
+
+// 仿真存储集合
+var SimulationMap = make(map[string]*wayside.VerifyMemory)
+
+// 仿真子id生成器
+var SimulationSubId = 0
+
+// 创建仿真对象
+func CreateSimulation(mapId int, mapData *graphicData.RtssGraphicStorage) string {
+	simulationId := createSimulationId(mapId)
+	verifyMemory := &wayside.VerifyMemory{
+		Status: &wayside.VerifyStatus{
+			SwitchStateMap: make(map[string]*state.SwitchState),
+		},
+	}
+	SimulationMap[simulationId] = verifyMemory
+	return simulationId
+}
+
+// 删除仿真对象
+func DestroySimulation(simulationId string) {
+	delete(SimulationMap, simulationId)
+}
+
+// 创建时生成仿真Id
+func createSimulationId(mapId int) string {
+	SimulationSubId++
+	return strconv.Itoa(mapId) + "-" + strconv.Itoa(SimulationSubId)
+}
+
+// 获取仿真列表
+func ListAllSimulations() []*wayside.VerifyMemory {
+	return nil
+}
+
+// 根据仿真id查找仿真实例
+func FindSimulation(simulationId string) *wayside.VerifyMemory {
+	return SimulationMap[simulationId]
+}

grpcproto/client.go

@@ -2,8 +2,8 @@ package apiproto
 
 import (
 	context "context"
-	"log"
 
+	"go.uber.org/zap"
 	grpc "google.golang.org/grpc"
 	"joylink.club/bj-rtsts-server/config"
 )
@@ -27,18 +27,21 @@ func Cli() CentrifugoApiClient {
 
 // 发布消息
 func PublishMsg(channalName string, data []byte) {
+	if len(data) == 0 {
+		return
+	}
 	resp, err := client.Publish(context.Background(), &PublishRequest{
 		Channel: channalName,
 		Data:    data,
 	})
 	if err != nil {
-		log.Fatalf("Transport level error: %v \n", err)
+		zap.S().Errorf("Transport level error: %v \n", err)
 	} else {
 		if resp.GetError() != nil {
 			respError := resp.GetError()
-			log.Fatalf("Publish msg[%s] error %d(%s)\n", channalName, respError.Code, respError.Message)
+			zap.S().Errorf("Publish msg[%s] error %d(%s)\n", channalName, respError.Code, respError.Message)
 		} else {
-			log.Println("Successfully published")
+			zap.S().Debugf("[%s] Successfully published", channalName)
 		}
 	}
 }

grpcproto/message.go

@@ -1,8 +1,9 @@
 package apiproto
 
 import (
-	"log"
 	"time"
+
+	"go.uber.org/zap"
 )
 
 // 消息服务
@@ -14,7 +15,7 @@ type IMsgServer interface {
 	getInterval() time.Duration
 
 	// 全量信息
-	allMsgData() []byte
+	allMsgData(params map[string]string) []byte
 
 	// 定时发送的消息
 	onTick() []TopicMsg
@@ -37,15 +38,18 @@ var serverExitChannelMap = make(map[string]chan bool)
 
 // 注册服务
 func RegisterMsgServer(server IMsgServer) {
+	if client == nil {
+		InitClient()
+	}
 	serverMap[server.getChannelName()] = &server
 	if server.getInterval() > 0 {
 		exitChannel := make(chan bool)
-		ticker := time.NewTicker(server.getInterval())
 		serverExitChannelMap[server.getChannelName()] = exitChannel
+		ticker := time.NewTicker(server.getInterval())
 		go func() {
 			defer func() {
 				if r := recover(); r != nil {
-					log.Fatalf("定时器发生错误，%v\n", r)
+					zap.S().Debug("定时器发生错误，%v\n", r)
 				}
 				ticker.Stop() // 意外退出时关闭定时器
 			}()
@@ -68,14 +72,15 @@ func RegisterMsgServer(server IMsgServer) {
 }
 
 // 用户初次进入系统后执行消息发送（这里可以使用对个人消息发送，但目前命名空间没有调试通）
-func Subscription() {
-	for key, server := range serverMap {
-		data := (*server).allMsgData()
-		if data != nil && len(data) > 0 {
-			PublishMsg(key, data)
-		}
-	}
-}
+// 这个有问题暂时注释掉
+//func Subscription() {
+//	for key, server := range serverMap {
+//		data := (*server).allMsgData()
+//		if data != nil && len(data) > 0 {
+//			PublishMsg(key, data)
+//		}
+//	}
+//}
 
 // 注销消息服务
 func UnRegisterMsgServer(key string) {
@@ -88,3 +93,8 @@ func UnRegisterMsgServer(key string) {
 		delete(serverMap, key)
 	}
 }
+
+// 获取消息服务
+func GetMsgServer(key string) *IMsgServer {
+	return serverMap[key]
+}

grpcproto/simulation_server.go

@@ -0,0 +1,84 @@
+package apiproto
+
+import (
+	"strconv"
+	"strings"
+	"time"
+
+	"joylink.club/bj-rtsts-server/ats/verify/protos/state"
+	"joylink.club/bj-rtsts-server/ats/verify/simulation"
+	"joylink.club/bj-rtsts-server/ats/verify/simulation/wayside"
+)
+
+type SimulationServer struct{}
+
+// 返回通道格式
+func (t *SimulationServer) getChannelName() string {
+	return "simulation-{sid}-devices-status"
+}
+
+// 消息运行间隔
+func (t *SimulationServer) getInterval() time.Duration {
+	return time.Second
+}
+
+// 返回所有数据
+func (t *SimulationServer) allMsgData(params map[string]string) []byte {
+	simId := params["SID"]
+	if simId == "" {
+		return nil
+	}
+	simulation := simulation.FindSimulation(simId)
+	if simulation == nil {
+		return nil
+	}
+	return []byte(generateTestState(simulation).String())
+}
+
+// 定时发送数据
+func (t *SimulationServer) onTick() []TopicMsg {
+	msgArr := make([]TopicMsg, len(simulation.SimulationMap))
+	i := 0
+	for k, v := range simulation.SimulationMap {
+		channelName := handlerChannelName(k, t.getChannelName())
+		data := generateTestState(v)
+		msgArr[i] = TopicMsg{
+			channalName: channelName,
+			data:        []byte(data.String()),
+		}
+		i++
+	}
+	return msgArr
+}
+
+// 处理订阅通道名称
+func handlerChannelName(sid string, format string) string {
+	return strings.Replace(format, "{sid}", sid, 1)
+}
+
+// 生成道岔测试状态数据
+func generateTestState(v *wayside.VerifyMemory) *state.PushedDevicesStatus {
+	switchMap := v.Status.SwitchStateMap
+	switchArr := make([]*state.SwitchState, 9)
+	if len(switchMap) == 0 {
+		for i := 0; i < 9; i++ {
+			switchArr[i] = &state.SwitchState{
+				Id:      strconv.Itoa(i),
+				Normal:  true,
+				Reverse: false,
+			}
+		}
+	} else {
+		i := 0
+		for _, v := range switchMap {
+			switchArr[i] = v
+			i++
+		}
+	}
+	return &state.PushedDevicesStatus{
+		All: true,
+		AllStatus: &state.AllDevicesStatus{
+			SwitchState: switchArr,
+		},
+	}
+}

grpcproto/test_server.go

@@ -14,7 +14,7 @@ func (t *TestServer) getInterval() time.Duration {
 	return time.Second
 }
 
-func (t *TestServer) allMsgData() []byte {
+func (t *TestServer) allMsgData(params map[string]string) []byte {
 	return []byte(`{"input": "hello world"}`)
 }