package sdk
|
|
import (
|
"fmt"
|
// "golang.org/x/sys/unix"
|
|
"taskpubsub/tasktag"
|
"taskpubsub/util"
|
|
"github.com/gogo/protobuf/proto"
|
|
"taskpubsub/logger"
|
|
"basic.com/pubsub/protomsg.git"
|
"basic.com/valib/deliver.git"
|
)
|
|
const (
|
postPush = "_1"
|
postPull = "_2"
|
)
|
|
var SocketManage = make(map[string]util.SocketContext)
|
var SdkMap = make(map[string]chan protomsg.SdkMessage)
|
|
var shm bool = false
|
|
var innerRecvTopic = []string{
|
"facedetect-sdk-no-track", //to sdk-no-track 以图搜图
|
}
|
|
var innerSendTopic = []string{
|
"facedetect-sdk-no-track", //to sdk-no-track 以图搜图
|
"virtual-faceextract-sdk-pull", //to web 以图搜图
|
}
|
|
func initInnerTopic() {
|
for _, sendTopic := range innerSendTopic {
|
createSdkSendServerAndListen(sendTopic)
|
}
|
|
for _, recvTopic := range innerRecvTopic {
|
createSdkRecvServerAndListen(recvTopic)
|
}
|
}
|
|
func Init(useShm bool) {
|
|
shm = useShm
|
|
logger.Info("============= init sdk info =====================")
|
for _, sdkid := range util.Sdklist { // 创建sdk server
|
createSdkTopicAndServer(sdkid)
|
logger.Info()
|
}
|
|
// 手动输入的主题
|
initInnerTopic()
|
|
// es
|
SdkMap["es"] = make(chan protomsg.SdkMessage)
|
logger.Info("create es channel: ")
|
go DealEsTopic()
|
|
go autoUpdateSdk(util.Sdkflag)
|
}
|
|
func createSdkTopicAndServer(sdkid string) {
|
createSdkSendServerAndListen(sdkid)
|
createSdkRecvServerAndListen(sdkid)
|
}
|
|
func deleteSdkTopicAndServer(id string) {
|
if _, isExist := SdkMap[id]; isExist { //存在
|
close(SdkMap[id])
|
delete(SdkMap, id)
|
logger.Info("删除主题 sdk: ", id)
|
}
|
|
if _, isExist := SocketManage[id]; isExist { //存在
|
SocketManage[id].Cancel()
|
delete(SocketManage, id)
|
logger.Info("删除server sdk: ", id)
|
}
|
}
|
|
func createSdkSendServerAndListen(id string) {
|
if _, isExist := SdkMap[id]; !isExist { //不存在
|
SdkMap[id] = make(chan protomsg.SdkMessage)
|
logger.Info("create", id)
|
}
|
|
url := fmt.Sprintf("ipc:///tmp/%s%s.ipc", id, postPush)
|
m := deliver.PushPull
|
if shm {
|
m = deliver.Shm
|
url = id + postPush
|
}
|
|
fmt.Println("SDK URL: ", url)
|
|
socket, err := util.NewSocketListen(int(m), url, shm)
|
if err != nil {
|
delete(SdkMap, id)
|
logger.Error(id, "create socket error!")
|
return
|
}
|
SocketManage[id] = socket
|
|
go Send(id, socket, SdkMap[id])
|
}
|
|
func createSdkRecvServerAndListen(id string) {
|
if _, isExist := SdkMap[id]; !isExist { //不存在
|
SdkMap[id] = make(chan protomsg.SdkMessage)
|
logger.Info("create", id)
|
}
|
|
url := fmt.Sprintf("ipc:///tmp/%s%s.ipc", id, postPull)
|
m := deliver.PushPull
|
if shm {
|
m = deliver.Shm
|
url = id + postPull
|
|
}
|
|
socket, err := util.NewSocketListen(int(m), url, shm)
|
|
if err != nil {
|
delete(SdkMap, id)
|
logger.Error(id, "create socket error!")
|
return
|
}
|
SocketManage[id] = socket
|
|
go Recv(socket)
|
}
|
|
//单独处理 es 主题的情况
|
func DealEsTopic() {
|
for {
|
select {
|
case <-SdkMap["es"]:
|
//logger.Info("es finanl sdk!")
|
}
|
}
|
}
|
|
//动态处理
|
func autoUpdateSdk(sdkflag chan bool) {
|
|
for _ = range sdkflag {
|
logger.Info("test autodelsdk")
|
var oldSdkList []string
|
for key := range SdkMap {
|
oldSdkList = append(oldSdkList, key)
|
}
|
|
newSdkList := util.Sdklist
|
|
// 手动添加的全部加上
|
for _, sendTopic := range innerSendTopic {
|
newSdkList = append(newSdkList, sendTopic)
|
}
|
for _, recvTopic := range innerRecvTopic {
|
newSdkList = append(newSdkList, recvTopic)
|
}
|
|
sdkListUpdate := util.Difference(oldSdkList, newSdkList)
|
logger.Info(sdkListUpdate)
|
|
for key, op := range sdkListUpdate {
|
if op == "add" {
|
createSdkTopicAndServer(key)
|
} else {
|
deleteSdkTopicAndServer(key)
|
}
|
}
|
}
|
}
|
|
//sdk数据 加工器
|
func ToSdkMsg(cid string, caddr string, taskid string, data []byte) protomsg.SdkMessage {
|
var sdkmsg = protomsg.SdkMessage{}
|
sdkmsg.Cid = cid
|
sdkmsg.Caddr = caddr
|
if val, ok := tasktag.TaskLabelMap.Load(taskid); !ok {
|
sdkmsg.Tasklab = nil
|
return sdkmsg
|
} else {
|
sdkmsg.Tasklab = val.(*protomsg.TaskLabel)
|
sdkmsg.Data = data
|
}
|
return sdkmsg
|
}
|
|
//sdk数据分发器
|
func GetSdkSendTopic(sdkmsg protomsg.SdkMessage) (sendTopic string) {
|
if int(sdkmsg.Tasklab.Index) < len(sdkmsg.Tasklab.Sdkinfos) {
|
sendTopic = sdkmsg.Tasklab.Sdkinfos[sdkmsg.Tasklab.Index].Ipcid
|
} else {
|
sendTopic = "es"
|
}
|
logger.Debug("分发的主题: ", sendTopic, "位置：", int(sdkmsg.Tasklab.Index)+1, "/", len(sdkmsg.Tasklab.Sdkinfos))
|
return
|
}
|
|
func Recv(socket util.SocketContext) {
|
var repsdkmsg = protomsg.SdkMessage{}
|
for {
|
select {
|
case <-socket.Context.Done():
|
logger.Info("socket close")
|
return
|
default:
|
if msg, err := socket.Sock.Recv(); err != nil {
|
continue
|
} else {
|
err = proto.Unmarshal(msg, &repsdkmsg)
|
if err != nil {
|
logger.Error("unmarshal error: ", err)
|
continue
|
}
|
repsdkmsg.Tasklab.Index++
|
//调用计算函数， 分发给下一个主题
|
nexttopic := GetSdkSendTopic(repsdkmsg)
|
SdkMap[nexttopic] <- repsdkmsg
|
}
|
}
|
}
|
}
|
|
func Send(sdkid string, socket util.SocketContext, in chan protomsg.SdkMessage) {
|
|
for {
|
select {
|
case <-socket.Context.Done():
|
logger.Info("socket is close")
|
return
|
case v, ok := <-in:
|
if ok {
|
data, err := v.Marshal()
|
if err != nil {
|
logger.Error("proto marshal error ", err)
|
continue
|
}
|
|
if err := socket.Sock.Send(data); err != nil {
|
logger.Error("failed send:sdkid=", sdkid)
|
continue
|
}
|
logger.Debug(sdkid, " send success: ", len(data))
|
} else {
|
logger.Debug(sdkid, " 主题关闭, 关闭send()")
|
return
|
}
|
}
|
}
|
}
|
