From ba96922f40f82d12e9a5f4cfa4d887e78f460125 Mon Sep 17 00:00:00 2001
From: zhangmeng <775834166@qq.com>
Date: 星期四, 26 十二月 2019 15:23:17 +0800
Subject: [PATCH] update

---
 camera/camera.go |  101 ++++++++++++++++++++++++++++++++++++++++++++------
 1 files changed, 89 insertions(+), 12 deletions(-)

diff --git a/camera/camera.go b/camera/camera.go
index ae7b05e..a542e7e 100644
--- a/camera/camera.go
+++ b/camera/camera.go
@@ -1,12 +1,14 @@
 package camera
 
 import (
-	"basic.com/valib/deliver.git"
-	"basic.com/valib/logger.git"
-	"basic.com/pubsub/protomsg.git"
-	"github.com/gogo/protobuf/proto"
+	"strings"
 	"taskpubsub/sdk"
 	"taskpubsub/util"
+
+	"basic.com/pubsub/protomsg.git"
+	"basic.com/valib/deliver.git"
+	"basic.com/valib/logger.git"
+	"github.com/gogo/protobuf/proto"
 )
 
 const (
@@ -16,17 +18,21 @@
 
 var SocketManage = make(map[string]util.SocketContext)
 
+var shm bool = false
+
 var innerRecvTopic = []string{
 	"virtual-faceextract-sdk-pull_2", //to web 浠ュ浘鎼滃浘
 }
 
 func initInnerTopic() {
-	for _, recvTopic := range innerRecvTopic {
-		createCameraRecvServerAndListen(recvTopic)
-	}
+	// for _, recvTopic := range innerRecvTopic {
+	// 	createCameraRecvServerAndListen(recvTopic)
+	// }
 }
 
-func Init() {
+func Init(useShm bool) {
+	shm = useShm
+
 	logger.Info("============ camera info ====================")
 	for _, cd := range util.CameraIds {
 		logger.Info(cd)
@@ -38,10 +44,20 @@
 		createCameraRecvServerAndListen(cam.Id)
 	}
 
+	//鏂囦欢鍒濆鍖�
+	for _, f := range util.FileArr {
+		logger.Info(f)
+	}
+	for _, f := range util.FileArr {
+		createCameraRecvServerAndListen(f.Id)
+	}
+
 	// 鎵嬪姩杈撳叆鐨勪富棰�
 	initInnerTopic()
 
 	go autoUpdateCamera(util.Cameraflag)
+
+	go autoUpdateFiles(util.Fileflag)
 }
 
 // camera 鎺ュ彈鏁版嵁
@@ -49,7 +65,15 @@
 	if _, isExist := SocketManage[id]; !isExist { //涓嶅瓨鍦�
 
 		url := "ipc:///tmp/" + id + ".ipc"
-		socket, err := util.NewSocketListen(int(deliver.PushPull), url)
+		m := deliver.PushPull
+		if shm {
+			m = deliver.Shm
+			url = id
+
+		}
+
+		logger.Info("CAMERA URL : ", url)
+		socket, err := util.NewSocketListen(int(m), url, shm)
 		if err != nil {
 			logger.Error("create socket error")
 			return
@@ -76,7 +100,9 @@
 		var oldcameras []string
 
 		for key := range SocketManage {
-			oldcameras = append(oldcameras, key)
+			if !strings.HasPrefix(key, util.File_Video_Id_Pre) && !strings.HasPrefix(key, util.File_Audio_Id_Pre) && !strings.HasPrefix(key, util.File_Img_Id_Pre) {
+				oldcameras = append(oldcameras, key)
+			}
 		}
 
 		var newcameras []string
@@ -102,7 +128,40 @@
 	}
 }
 
+//鍔ㄦ�佸鐞嗘枃浠�
+func autoUpdateFiles(fileflag chan bool) {
+
+	for _ = range fileflag {
+		logger.Info("do auto del fileflag")
+		var oldIds []string
+
+		for key := range SocketManage {
+			if strings.HasPrefix(key, util.File_Video_Id_Pre) || strings.HasPrefix(key, util.File_Audio_Id_Pre) || strings.HasPrefix(key, util.File_Img_Id_Pre) {
+				oldIds = append(oldIds, key)
+			}
+		}
+
+		var newIds []string
+		for _, f := range util.FileArr {
+			newIds = append(newIds, f.Id)
+		}
+
+		fileListUpdate := util.Difference(oldIds, newIds)
+		logger.Info(fileListUpdate)
+
+		for key, op := range fileListUpdate {
+			if op == "add" {
+				createCameraRecvServerAndListen(key)
+			} else {
+				deleteCameraRecvServer(key)
+			}
+		}
+	}
+}
+
 func Recv(socket util.SocketContext) {
+
+	tryCount := 0
 
 	var recvmessage []byte
 	var imagemsg protomsg.Image
@@ -110,11 +169,26 @@
 	for {
 		select {
 		case <-socket.Context.Done():
+			socket.Sock.Close()
 			logger.Error("listen recv quit")
 			return
 		default:
 			if recvmessage, err = socket.Sock.Recv(); err != nil {
-				//logger.Error("[camera] err is: ", err)
+
+				// tryCount++
+				// socket = util.MaybeRestartSocket(socket, &tryCount)
+
+				if socket.UseSHM {
+					if tryCount > util.ShmMaxTryCount {
+						socket.Sock.Close()
+						socket, err = util.NewSocketListen(socket.Mode, socket.URL, socket.UseSHM)
+						logger.Info("=========CAMERA RECV SHM ERROR=========")
+						logger.Info("CAMERA RECV FROM:", socket.URL, " ERROR ", util.ShmMaxTryCount, " TIMES SO RESTART IT")
+						tryCount = 0
+						continue
+					}
+					tryCount++
+				}
 				continue
 			}
 
@@ -134,7 +208,7 @@
 			} else {
 				taskIDs := GetAllTaskByID(imagemsg.Cid)
 				for _, taskID := range taskIDs {
-					//logger.Info("id: ", imagemsg.Cid, " taskid: ", taskID)
+					logger.Info("id: ", imagemsg.Cid, " taskid: ", taskID)
 					doTaskList(imagemsg.Cid, "", taskID, recvmessage)
 				}
 			}
@@ -147,6 +221,9 @@
 	for _, camsingle := range util.CameraTasks {
 		if cid == camsingle.Camera.Id {
 			for _, tasksingle := range camsingle.Tasks {
+				if !tasksingle.Enable {
+					continue
+				}
 				tasks = append(tasks, tasksingle.Taskid)
 			}
 			return

--
Gitblit v1.8.0