VideoStream/main.go

package main

import (
	"VideoStream/dao"
	"VideoStream/handler"
	"VideoStream/proto"
	"VideoStream/service"
	"VideoStream/worker"
	"fmt"
	"github.com/gin-gonic/gin"
	"github.com/golang-jwt/jwt"
	"github.com/robfig/cron/v3"
	"gocv.io/x/gocv"
	"log"
	"runtime"
	"strconv"
	"strings"
	"sync"
)

var configPath string
var logCount int

func main() {
	fmt.Println("start server")
	gin.SetMode(gin.ReleaseMode) //设置为debug模式
	r := gin.Default()
	//数据库初始
	err0 := dao.Init()
	defer dao.Close()
	if err0 != nil {
		panic("failed to connect database:" + err0.Error())
	}
	err0 = worker.InitRedis()
	defer worker.CloseRedis()
	if err0 != nil {
		panic("failed to connect redis:" + err0.Error())
	}
	r.Use(handler.CorsHandler())
	r.Use(JWTAuthMiddleware()) // 使用 JWT 认证中间件
	handler.SetUpToolGroup(r)  // Tool
	err := worker.InitRedis()
	if err != nil {
		panic("failed to connect redis:" + err.Error())
	}
	//定时任务
	c := cron.New(cron.WithSeconds())
	// 添加每 10 秒执行一次的任务
	_, err2 := c.AddFunc("@every 10s", myTask)
	if err2 != nil {
		log.Fatal("添加定时任务失败: ", err2)
	}
	c.Start()
	fmt.Println("定时任务已启动")
	initDeviceGettingStatus()
	err3 := r.Run(":" + proto.Config.SERVER_PORT)
	if err3 != nil {
		panic("failed to run server:" + err3.Error())
	}
}

func init() {
	//读取配置文件
	//文件地址/home/videoplayer/vp.conf
	os := runtime.GOOS
	if os == "windows" {
		configPath = "D:/File/vp_stream.conf"
	} else {
		configPath = "/home/videoplayer/vp_stream.conf"
	}
	logCount = 0
	//读取配置文件
	err := proto.ReadConfig(configPath)
	if err != nil {
		panic("failed to read config file:" + err.Error())
	}
	service.DeviceFrameCount = make(map[int]int)
	for _, device := range proto.Config.DeviceInfo {
		//service.DeviceRWMap[device.ID] = sync.RWMutex{}
		//service.DeviceCurrentFrameMap[device.ID] = gocv.NewMat()
		//service.DeviceIsGettingFrame[device.ID] = false
		service.DeviceRWMap.Store(device.ID, &sync.RWMutex{})
		service.DeviceCurrentFrameMap.Store(device.ID, gocv.NewMat())
		service.DeviceFrameCount[device.ID] = 0
		service.DeviceIsGettingFrame.Store(device.ID, false)
	}
}

func myTask() {
	ReadConfigAndSetSystem()
}

func initDeviceGettingStatus() {
	for _, device := range proto.Config.DeviceInfo {
		if device.NextStop == false {
			worker.SetRedis(fmt.Sprintf("device_%d_is_getting", device.ID), "false")
		}
		if device.NextStop == true {
			worker.DelRedis(fmt.Sprintf("device_%d_is_getting", device.ID))
		}
	}

}

func ReadConfigAndSetSystem() {
	logCount++
	//configPath := "/home/videoplayer/vp_stream.conf"
	//读取配置文件
	err := proto.ReadConfig(configPath)
	if err != nil {
		panic("failed to read config file:" + err.Error())
	}
	//检测是否需要获取设备流,如果需要则开启
	for _, device := range proto.Config.DeviceInfo {
		//isGet, ok := service.DeviceIsGettingFrame.Load(device.ID)
		//isGet_ := isGet.(bool)
		isGetting := worker.GetRedis(fmt.Sprintf("device_%d_is_getting", device.ID))
		if isGetting == "" {
			//说明没有这个设备，需初始化添加
			service.DeviceRWMap.Store(device.ID, &sync.RWMutex{})
			service.DeviceCurrentFrameMap.Store(device.ID, gocv.NewMat())
			service.DeviceFrameCount[device.ID] = 0
			service.DeviceIsGettingFrame.Store(device.ID, false)
			worker.SetRedis(fmt.Sprintf("device_%d_is_getting", device.ID), "false")
		}
		if isGetting == "false" && device.NextStop == false { //如果设备流已经停止且不暂停，则开启
			switch device.ID {
			case 1:
				service.Device1CurrentFrame = gocv.NewMat()
			case 50:
				service.Device50CurrentFrame = gocv.NewMat()
			case 73:
				service.Device73CurrentFrame = gocv.NewMat()
			}
			go service.GetVideoStream(device.ID)
			log.Printf("device:%d has started!\n", device.ID)
		}
	}
	if logCount%3600 == 0 {
		log.Printf("每%d秒执行一次,当前设备：%v", logCount*10, proto.Config.DeviceInfo)
	}
}

func JWTAuthMiddleware() gin.HandlerFunc {
	return func(c *gin.Context) {
		// 从请求头中获取 JWT 令牌
		tokenString := c.Request.Header.Get("token")

		//请求方式为get时，从url中获取token
		if tokenString == "" {
			tokenString = c.Query("token")
		}

		//如果请求为login或register，则不需要验证token
		for k, _ := range proto.Url_map {
			if strings.Contains(c.Request.URL.Path, k) {
				c.Next()
				return
			}
		}
		if tokenString == "" {
			//c.AbortWithStatus(200)
			c.JSON(200, gin.H{
				"message": "Unauthorized",
				"error":   "token is empty",
				"code":    3,
			})
			return
		}
		if proto.Config.TOKEN_USE_REDIS {
			redisToken := worker.GetRedis(tokenString)
			if redisToken == "" {
				c.AbortWithStatus(200)
				c.JSON(200, gin.H{
					"message": "NOT_LOGIN",
					"error":   "server token is empty",
					"code":    3,
				})
				return
			}
		}
		//查看token是否在超级token中
		if worker.IsContainSet("super_permission_tokens", tokenString) {
			s_id := c.Request.Header.Get("super_id")
			if s_id == "" {
				c.AbortWithStatus(200)
				c.JSON(200, gin.H{
					"message": "NOT_LOGIN",
					"error":   "super_id is empty",
					"code":    3,
				})
				return
			}
			id, _ := strconv.Atoi(s_id)
			//查看s_id类型
			c.Set("id", id)
			c.Next()
			return
		}

		// 使用加密secret 解析 JWT 令牌
		token, err := jwt.Parse(tokenString, func(token *jwt.Token) (interface{}, error) {
			return proto.SigningKey, nil
		})

		// 验证令牌
		if err != nil || !token.Valid {
			c.AbortWithStatus(200)
			c.JSON(200, gin.H{
				"message": "NOT_LOGIN",
				"error":   "Invalid token",
				"code":    4,
			})
			return
		}
		userId := int(token.Claims.(jwt.MapClaims)["id"].(float64))
		// 将用户信息添加到上下文中
		c.Set("id", userId)
		// 继续处理请求
		c.Next()
	}
}