Compare commits
No commits in common. "1e6a0996a053f3b88a217748a123d44995d68049" and "d69e02115fb1b1f228276d22ba4345e53d6b875f" have entirely different histories.
1e6a0996a0
...
d69e02115f
|
|
@ -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,78 +65,48 @@ 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"
|
||||
}
|
||||
|
||||
if err := c.ShouldBind(&req); err != nil {
|
||||
c.JSON(400, gin.H{"error": err.Error()})
|
||||
return
|
||||
} else {
|
||||
//查看id是否存在
|
||||
index := -1
|
||||
for _, device_ := range proto.Config.DeviceInfo {
|
||||
if device_.ID == deviceID {
|
||||
index = deviceID
|
||||
for _, device := range proto.Config.DeviceInfo {
|
||||
if device.ID == req.ID {
|
||||
index = req.ID
|
||||
break
|
||||
}
|
||||
}
|
||||
if index == -1 {
|
||||
c.JSON(400, gin.H{"error": "id config not exist"})
|
||||
c.JSON(400, gin.H{"error": "id not exist"})
|
||||
return
|
||||
}
|
||||
//查看key是否正确
|
||||
if key != rKey {
|
||||
if req.Key != "123456" {
|
||||
c.JSON(400, gin.H{"error": "key error"})
|
||||
return
|
||||
}
|
||||
//设备流
|
||||
c.Stream(func(w io.Writer) bool {
|
||||
var count int
|
||||
frame := gocv.NewMat()
|
||||
defer func() {
|
||||
//关闭帧
|
||||
err4 := frame.Close()
|
||||
if err4 != nil {
|
||||
log.Printf("device:%d frame close err:%v", deviceID, err4)
|
||||
}
|
||||
}()
|
||||
errCount := 0
|
||||
for {
|
||||
if errCount > 10 {
|
||||
log.Printf("stream device:%d errCount > 10", deviceID)
|
||||
return false
|
||||
}
|
||||
cnt := service.GetDeviceCurrentFrameV2(&frame, deviceID)
|
||||
frame, cnt := service.GetDeviceCurrentFrame(req.ID)
|
||||
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
|
||||
}
|
||||
img, err := gocv.IMEncode(".jpg", frame)
|
||||
frame_ := img.GetBytes()
|
||||
|
||||
_, err = w.Write([]byte("--frame\r\nContent-Type: image/jpeg\r\n\r\n"))
|
||||
|
|
@ -157,6 +127,8 @@ func GetVideoStream(c *gin.Context) {
|
|||
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))
|
||||
if isGetting != "true" {
|
||||
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)
|
||||
is_get, ok := service.DeviceIsGettingFrame.Load(device.ID)
|
||||
if !ok || is_get == false {
|
||||
//直接返回
|
||||
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 {
|
||||
|
|
|
|||
35
main.go
35
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)
|
||||
//isGet_ := isGet.(bool)
|
||||
isGetting := worker.GetRedis(fmt.Sprintf("device_%d_is_getting", device.ID))
|
||||
if isGetting == "" {
|
||||
is_get, ok := service.DeviceIsGettingFrame.Load(device.ID)
|
||||
if !ok {
|
||||
//说明没有这个设备，需初始化添加
|
||||
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.Printf("每%d秒执行一次,当前设备：%v", logCount*10, proto.Config.DeviceInfo)
|
||||
}
|
||||
log.Println("每10秒执行一次,当前设备：", proto.Config.DeviceInfo)
|
||||
}
|
||||
|
||||
func JWTAuthMiddleware() gin.HandlerFunc {
|
||||
|
|
|
|||
|
|
@ -109,25 +109,17 @@ func GetDeviceCurrentFrameV2(frame *gocv.Mat, deviceId int) int {
|
|||
}
|
||||
mutex.RLock()
|
||||
defer mutex.RUnlock()
|
||||
var frame_ gocv.Mat
|
||||
//获取当前帧
|
||||
switch deviceId {
|
||||
case 1:
|
||||
if (*frame).Empty() {
|
||||
*frame = gocv.NewMatWithSize(Device1CurrentFrame.Rows(), Device1CurrentFrame.Cols(), Device1CurrentFrame.Type())
|
||||
frameIface, ok := DeviceCurrentFrameMap.Load(deviceId)
|
||||
if !ok {
|
||||
return -1
|
||||
}
|
||||
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))
|
||||
if isGetting == "true" || isGetting == "" {
|
||||
log.Printf("设备:%d 正在运行,isGetting:%s", id, isGetting)
|
||||
is_get, ok := DeviceIsGettingFrame.Load(id)
|
||||
if !ok {
|
||||
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))
|
||||
if isGetting == "true" {
|
||||
log.Printf("设备:%d 正在运行,isGetting:%s", id, isGetting)
|
||||
is_get, ok = DeviceIsGettingFrame.Load(id)
|
||||
if is_get == true {
|
||||
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)
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue