handler/tool.go

@@ -34,7 +34,7 @@ func SetUpToolGroup(router *gin.Engine) {
 }
 
 // 跨域访问：cross  origin resource share
-func CorsHandler() gin.HandlerFunc {
+func CrosHandler() gin.HandlerFunc {
 	return func(context *gin.Context) {
 		//method := context.Request.Method
 		context.Writer.Header().Set("Access-Control-Allow-Origin", "*")
@@ -65,98 +65,70 @@ type videoStreamReq struct {
 }
 
 func GetVideoStream(c *gin.Context) {
+	var req videoStreamReq
 	id, _ := c.Get("id")
 	id1 := id.(int)
-	deviceIDSTR := c.Query("id")
-	deviceID, err := strconv.Atoi(deviceIDSTR)
-	if err != nil {
-		c.JSON(400, gin.H{"error": "device_id error"})
-		return
-	}
-	key := c.Query("key")
 	//校验权限
-	device := service.GetDevice(deviceID, id1)
+	device := service.GetDevice(req.ID, id1)
 	if device.ID == 0 {
 		c.JSON(400, gin.H{"error": "device not exist"})
 		return
 	}
-	rKey := worker.GetRedis("video_stream_get_stream_key")
-	if rKey == "" {
-		rKey = "123456"
-	}
 
-	//查看id是否存在
+	if err := c.ShouldBind(&req); err != nil {
-	index := -1
+		c.JSON(400, gin.H{"error": err.Error()})
-	for _, device_ := range proto.Config.DeviceInfo {
+		return
-		if device_.ID == deviceID {
+	} else {
-			index = deviceID
+		//查看id是否存在
-			break
+		index := -1
+		for _, device := range proto.Config.DeviceInfo {
+			if device.ID == req.ID {
+				index = req.ID
+				break
+			}
 		}
-	}
+		if index == -1 {
-	if index == -1 {
+			c.JSON(400, gin.H{"error": "id not exist"})
-		c.JSON(400, gin.H{"error": "id config not exist"})
+			return
-		return
+		}
-	}
+		//查看key是否正确
-	//查看key是否正确
+		if req.Key != "123456" {
-	if key != rKey {
+			c.JSON(400, gin.H{"error": "key error"})
-		c.JSON(400, gin.H{"error": "key error"})
+			return
-		return
+		}
-	}
+		//设备流
-	//设备流
+		c.Stream(func(w io.Writer) bool {
-	c.Stream(func(w io.Writer) bool {
+			var count int
-		var count int
+			for {
-		frame := gocv.NewMat()
+				frame, cnt := service.GetDeviceCurrentFrame(req.ID)
-		defer func() {
+				if cnt == count {
-			//关闭帧
+					time.Sleep(50 * time.Millisecond)
-			err4 := frame.Close()
+					continue
-			if err4 != nil {
+				}
-				log.Printf("device:%d frame close err:%v", deviceID, err4)
+				//gocv.Matrix转为jpeg
-			}
+				img, err := gocv.IMEncode(".jpg", frame)
-		}()
+				frame_ := img.GetBytes()
-		errCount := 0
-		for {
-			if errCount > 10 {
-				log.Printf("stream device:%d errCount > 10", deviceID)
-				return false
-			}
-			cnt := service.GetDeviceCurrentFrameV2(&frame, deviceID)
-			if cnt == count {
-				time.Sleep(50 * time.Millisecond)
-				errCount++
-				continue
-			}
-			if frame.Empty() {
-				log.Printf("stream device:%d frame is empty", deviceID)
-				time.Sleep(100 * time.Millisecond)
-				errCount++
-				continue
-			}
-			//gocv.Matrix转为jpeg
-			img, err2 := gocv.IMEncode(".jpg", frame)
-			if err2 != nil {
-				log.Printf("stream img encode err:%v", err2)
-				return false
-			}
-			frame_ := img.GetBytes()
 
-			_, err = w.Write([]byte("--frame\r\nContent-Type: image/jpeg\r\n\r\n"))
+				_, err = w.Write([]byte("--frame\r\nContent-Type: image/jpeg\r\n\r\n"))
-			if err != nil {
+				if err != nil {
-				fmt.Printf("写入头部信息错误: %v\n", err)
+					fmt.Printf("写入头部信息错误: %v\n", err)
-				return false
+					return false
+				}
+				_, err = w.Write(frame_)
+				if err != nil {
+					fmt.Printf("写入帧数据错误: %v\n", err)
+					return false
+				}
+				_, err = w.Write([]byte("\r\n"))
+				if err != nil {
+					fmt.Printf("写入帧结束标记错误: %v\n", err)
+					return false
+				}
+				time.Sleep(50 * time.Millisecond) // 控制帧率，模拟每秒约20帧，可按实际调整
 			}
-			_, err = w.Write(frame_)
+		})
-			if err != nil {
+
-				fmt.Printf("写入帧数据错误: %v\n", err)
+	}
-				return false
-			}
-			_, err = w.Write([]byte("\r\n"))
-			if err != nil {
-				fmt.Printf("写入帧结束标记错误: %v\n", err)
-				return false
-			}
-			time.Sleep(50 * time.Millisecond) // 控制帧率，模拟每秒约20帧，可按实际调整
-		}
-	})
 }
 
 // 发送实时视频流
@@ -171,13 +143,13 @@ func GetRealTimeImage(c *gin.Context) {
 		return
 	}
 	//查看设备是否在获取
-	isGetting := worker.GetRedis(fmt.Sprintf("device_%d_is_getting", device.ID))
+	is_get, ok := service.DeviceIsGettingFrame.Load(device.ID)
-	if isGetting != "true" {
+	if !ok || is_get == false {
-		c.JSON(http.StatusOK, gin.H{"code": 4, "message": "device is not getting or not exist"})
+		//直接返回
-		log.Printf("device_id:%d is not getting or not exist", deviceIdInt)
+		c.JSON(http.StatusOK, gin.H{"code": 4, "message": "device not getting frame or not exist"})
+		log.Printf("device:%d not found", device.ID)
 		return
 	}
-
 	ws, err := upgrader.Upgrade(c.Writer, c.Request, nil)
 	if err != nil {
 		log.Printf("connect wss err:%v", err)
@@ -198,10 +170,9 @@ func subscribeAndHandleMessagesV3(ws *websocket.Conn, device_id int) {
 	count := 0
 	//定时器，发送计数器
 	t_count := 0
-	img := gocv.NewMat()
 	for {
 		//从service获取当前帧
-		c := service.GetDeviceCurrentFrameV2(&img, device_id)
+		img, c := service.GetDeviceCurrentFrameV3(device_id)
 		if c != count {
 			//将img转[]byte
 			if img.Empty() {
@@ -223,6 +194,10 @@ func subscribeAndHandleMessagesV3(ws *websocket.Conn, device_id int) {
 					break
 				}
 				c = count
+				err4 := img.Close()
+				if err4 != nil {
+					log.Printf("close img err:%v", err)
+				}
 			}
 		} else {
 			//每秒发送一次心跳检测
@@ -240,11 +215,6 @@ func subscribeAndHandleMessagesV3(ws *websocket.Conn, device_id int) {
 		time.Sleep(100 * time.Millisecond)
 		t_count++
 	}
-	// 关闭img
-	err := img.Close()
-	if err != nil {
-		log.Printf("device:%d img close err:%v", device_id, err)
-	}
 
 	// 查看是否还有其他连接，没有则设置 is_play 为 0
 	if worker.IsContainKey(online_conn_key) == false {

main.go

@@ -19,11 +19,10 @@ import (
 )
 
 var configPath string
-var logCount int
 
 func main() {
 	fmt.Println("start server")
-	gin.SetMode(gin.ReleaseMode) //设置为debug模式
+	gin.SetMode(gin.DebugMode) //设置为debug模式
 	r := gin.Default()
 	//数据库初始
 	err0 := dao.Init()
@@ -36,7 +35,7 @@ func main() {
 	if err0 != nil {
 		panic("failed to connect redis:" + err0.Error())
 	}
-	r.Use(handler.CorsHandler())
+	r.Use(handler.CrosHandler())
 	r.Use(JWTAuthMiddleware()) // 使用 JWT 认证中间件
 	handler.SetUpToolGroup(r)  // Tool
 	err := worker.InitRedis()
@@ -52,7 +51,6 @@ func main() {
 	}
 	c.Start()
 	fmt.Println("定时任务已启动")
-	initDeviceGettingStatus()
 	err3 := r.Run(":" + proto.Config.SERVER_PORT)
 	if err3 != nil {
 		panic("failed to run server:" + err3.Error())
@@ -68,7 +66,6 @@ func init() {
 	} else {
 		configPath = "/home/videoplayer/vp_stream.conf"
 	}
-	logCount = 0
 	//读取配置文件
 	err := proto.ReadConfig(configPath)
 	if err != nil {
@@ -90,20 +87,7 @@ 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)
@@ -112,33 +96,26 @@ func ReadConfigAndSetSystem() {
 	}
 	//检测是否需要获取设备流,如果需要则开启
 	for _, device := range proto.Config.DeviceInfo {
-		//isGet, ok := service.DeviceIsGettingFrame.Load(device.ID)
+		is_get, ok := service.DeviceIsGettingFrame.Load(device.ID)
-		//isGet_ := isGet.(bool)
+		if !ok {
-		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 { //如果设备流已经停止且不暂停，则开启
+		if is_get == 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.Println("每10秒执行一次,当前设备：", proto.Config.DeviceInfo)
-		log.Printf("每%d秒执行一次,当前设备：%v", logCount*10, proto.Config.DeviceInfo)
-	}
 }
 
 func JWTAuthMiddleware() gin.HandlerFunc {

service/tool.go

@@ -109,25 +109,17 @@ func GetDeviceCurrentFrameV2(frame *gocv.Mat, deviceId int) int {
 	}
 	mutex.RLock()
 	defer mutex.RUnlock()
+	var frame_ gocv.Mat
 	//获取当前帧
-	switch deviceId {
+	frameIface, ok := DeviceCurrentFrameMap.Load(deviceId)
-	case 1:
+	if !ok {
-		if (*frame).Empty() {
+		return -1
-			*frame = gocv.NewMatWithSize(Device1CurrentFrame.Rows(), Device1CurrentFrame.Cols(), Device1CurrentFrame.Type())
-		}
-		Device1CurrentFrame.CopyTo(frame)
-	case 50:
-		if (*frame).Empty() {
-			*frame = gocv.NewMatWithSize(Device50CurrentFrame.Rows(), Device50CurrentFrame.Cols(), Device50CurrentFrame.Type())
-		}
-		Device50CurrentFrame.CopyTo(frame)
-	case 73:
-		if (*frame).Empty() {
-			*frame = gocv.NewMatWithSize(Device73CurrentFrame.Rows(), Device73CurrentFrame.Cols(), Device73CurrentFrame.Type())
-		}
-		Device73CurrentFrame.CopyTo(frame)
-
 	}
+	frame_, ok = frameIface.(gocv.Mat)
+	if !ok {
+		log.Printf("DeviceCurrentFrameMap 存储的不是 gocv.Mat 类型，device_id: %d \n", deviceId)
+	}
+	*frame = frame_
 	frameCount, ok := DeviceFrameCount[deviceId]
 	if !ok {
 		return -1
@@ -169,9 +161,7 @@ func GetDeviceCurrentFrameV3(deviceId int) (gocv.Mat, int) {
 		Device50CurrentFrame.CopyTo(&frame)
 		//查看帧状态
 		//log.Printf("frame:%v,Device50CurrentFrame:%v\n", frame.Empty(), Device50CurrentFrame.Empty())
-	case 73:
+
-		frame = gocv.NewMatWithSize(Device73CurrentFrame.Rows(), Device73CurrentFrame.Cols(), Device73CurrentFrame.Type())
-		Device73CurrentFrame.CopyTo(&frame)
 	}
 
 	frameCount, ok := DeviceFrameCount[deviceId]
@@ -311,9 +301,13 @@ func GetVideoStream(id int) {
 			log.Printf("设备:%d 错误: %v\n", id, err)
 		}
 	}()
-	isGetting := worker.GetRedis(fmt.Sprintf("device_%d_is_getting", id))
+	is_get, ok := DeviceIsGettingFrame.Load(id)
-	if isGetting == "true" || isGetting == "" {
+	if !ok {
-		log.Printf("设备:%d 正在运行,isGetting:%s", id, isGetting)
+		log.Printf("device: %d  not found", id)
+		return
+	}
+	if is_get == true {
+		log.Printf("device: %d is running!", id)
 		return
 	}
 	for {
@@ -332,9 +326,9 @@ func GetVideoStream(id int) {
 			log.Printf("device: %d not found", id)
 			break
 		}
-		isGetting = worker.GetRedis(fmt.Sprintf("device_%d_is_getting", id))
+		is_get, ok = DeviceIsGettingFrame.Load(id)
-		if isGetting == "true" {
+		if is_get == true {
-			log.Printf("设备:%d 正在运行,isGetting:%s", id, isGetting)
+			log.Printf("for device:%d is running!", id)
 			break
 		}
 
@@ -345,11 +339,9 @@ func GetVideoStream(id int) {
 		//设置设备控制信息
 		status := Get(device.Control)
 		DeviceIsGettingFrame.Store(id, true)
-		worker.SetRedis(fmt.Sprintf("device_%d_is_getting", id), "true")
 		log.Printf("device: %d set control info status: %d", device.ID, status)
 		getVideoFrame(device)
 		DeviceIsGettingFrame.Store(id, false)
-		worker.SetRedis(fmt.Sprintf("device_%d_is_getting", id), "false")
 		//等待1s
 		time.Sleep(1 * time.Second)
 	}