by 192.168.1.124

龙赣华

2019-06-05 d7ee5b3444e7681cfb4574681d54156e148c34da

by 192.168.1.124

 camera/camera.go

@@ -37,6 +37,7 @@
    }

    go CreateCamera(Initchannel)
    go AutoDelCamera(util.Cameraflag)

    for _, cam := range util.CameraIds {
        Initchannel <- cam.Id
@@ -57,6 +58,45 @@
            go func(cid string, sock SocketContext) {
                Recv(cid, sock)
            }(id, socketlisten)
        }
    }
}

//动态处理
func AutoDelCamera(cameraflag chan bool) {

    for _ = range cameraflag {
        fmt.Println("test autodelcameraflag")
        var oldcamera []string

        SocketManage.Range(func(k, v interface{}) bool {
            if str, ok := k.(string); ok {
                oldcamera = append(oldcamera, str)
            }
            return true
        })

        var newcamera []string

        for _, value := range util.CameraIds {
            newcamera = append(newcamera, value.Id)
        }

        cameraChanDel := util.Difference(oldcamera, newcamera)
        fmt.Println(cameraChanDel)

        for key, op := range cameraChanDel {
            if op == "add" {
                Initchannel <- key
            } else {
                if sock, ok := SocketManage.Load(key); ok {
                    if socket, sok := sock.(SocketContext); sok {
                        socket.Cancel()
                        SocketManage.Delete(key)
                    }
                }
                fmt.Println("删除camera server : ", key)
            }
        }
    }
}
@@ -122,7 +162,7 @@
    //  数据加工(打标签)
    sdkmsg := sdk.SdkData(cid, taskid, data)
    if sdkmsg.Tasklab == nil {
        fmt.Println("cid:%s 没有任务%s", cid, taskid)
        fmt.Printf("cid:%s 没有任务%s\n", cid, taskid)
        return
    }


 go.mod

@@ -3,18 +3,17 @@
go 1.12

require (
    basic.com/dbapi.git v0.0.0-20190523103034-c0d33072098b
    basic.com/dbapi.git v0.0.0-20190531051326-e49473afa5e9
    basic.com/pubsub/protomsg.git v0.0.0-20190530084829-0ea842491a96
    basic.com/r/dbapi.git v0.0.0-20190523103034-c0d33072098b
    basic.com/r/dbapi.git v0.0.0-20190531051326-e49473afa5e9
    basic.com/valib/deliver.git v0.0.0-20190529080650-3e64847c9bce
    basic.com/valib/gopherdiscovery.git v0.0.0-20190530055034-f8c281d054f6
    github.com/gogo/protobuf v1.2.1
    github.com/golang/protobuf v1.3.1
    github.com/gorilla/websocket v1.4.0 // indirect
    github.com/smartystreets/goconvey v0.0.0-20190330032615-68dc04aab96a // indirect
    github.com/smartystreets/goconvey v0.0.0-20190330032615-68dc04aab96a
    github.com/tmthrgd/go-sem v0.0.0-20160607101025-0214dbf53877 // indirect
    github.com/tmthrgd/go-shm v0.0.0-20170117044846-90afcfcd5ee9 // indirect
    github.com/tmthrgd/shm-go v0.0.0-20170130075737-7207ca97b290 // indirect
    golang.org/x/net v0.0.0-20190522155817-f3200d17e092 // indirect
    nanomsg.org/go-mangos v1.4.0 // indirect
    golang.org/x/net v0.0.0-20190522155817-f3200d17e092
    nanomsg.org/go-mangos v1.4.0
)

 go.sum

@@ -1,21 +1,32 @@
basic.com/dbapi.git v0.0.0-20190523103034-c0d33072098b h1:gzr51BWE821BzyhRb0iiP5Wu/yXTkbfcz0BkzfjacMs=
basic.com/dbapi.git v0.0.0-20190523103034-c0d33072098b/go.mod h1:eDXPnxaz6jZPDvBSk7ya7oSASWPCuUEgRTJCjsfKt/Q=
basic.com/dbapi.git v0.0.0-20190531051326-e49473afa5e9 h1:1HIG2sEEYVUKL7nyJvURj/p24JpDwFwKE/XtatPsXVQ=
basic.com/dbapi.git v0.0.0-20190531051326-e49473afa5e9/go.mod h1:eDXPnxaz6jZPDvBSk7ya7oSASWPCuUEgRTJCjsfKt/Q=
basic.com/pubsub/protomsg.git v0.0.0-20190530084829-0ea842491a96 h1:7nkipxWDbIK4wRbLCZeUoGxzdEIxZFumSTM6xhfWiWM=
basic.com/pubsub/protomsg.git v0.0.0-20190530084829-0ea842491a96/go.mod h1:un5NV5VWQoblVLZfx1Rt5vyLgwR0jI92d3VJhfrJhWU=
basic.com/r/dbapi.git v0.0.0-20190523103034-c0d33072098b h1:gIfKQeYTnO/y6mfrhx1ZPAyk0kgeocma9DXsgRdimoI=
basic.com/r/dbapi.git v0.0.0-20190523103034-c0d33072098b/go.mod h1:5ia9iOg9Q8kvHoxo9rAOMF++ws7h1Mvj8b4L9C4c2MA=
basic.com/r/dbapi.git v0.0.0-20190531051326-e49473afa5e9 h1:P6F1NSgD5VkWNqrgwT/6UU68tTGiUCULrFY/MxHlTk0=
basic.com/r/dbapi.git v0.0.0-20190531051326-e49473afa5e9/go.mod h1:5ia9iOg9Q8kvHoxo9rAOMF++ws7h1Mvj8b4L9C4c2MA=
basic.com/valib/deliver.git v0.0.0-20190529080650-3e64847c9bce h1:/D6k+FVN1sMqLz6tMlsIl9bKwE2Mpc5d4QfPh0y4DSQ=
basic.com/valib/deliver.git v0.0.0-20190529080650-3e64847c9bce/go.mod h1:bkYiTUGzckyNOjAgn9rB/DOjFzwoSHJlruuWQ6hu6IY=
basic.com/valib/gopherdiscovery.git v0.0.0-20190530055034-f8c281d054f6 h1:0srMU/gU5qyX726Jvi28dXN8StnmlVGRbj+SzI/8OVs=
basic.com/valib/gopherdiscovery.git v0.0.0-20190530055034-f8c281d054f6/go.mod h1:CQ+UJyZV8MRzwwckncdUDu6/RDTKAzSIPCxc9tFcwPs=
basic.com/valib/gopherdiscovery.git v0.0.0-20190530092835-5bc9943400c2 h1:k5NHm1Ch8cy+YNRxQuTEnRsd+VSXpqDK/UDUdjwG5qY=
basic.com/valib/gopherdiscovery.git v0.0.0-20190530092835-5bc9943400c2/go.mod h1:CQ+UJyZV8MRzwwckncdUDu6/RDTKAzSIPCxc9tFcwPs=
basic.com/valib/gopherdiscovery.git v0.0.0-20190531081358-bde1bf71186f h1:n1IcA2ZiLzKcPUOBA09CUlgyzAPEflfjoTf/gJOHgp8=
basic.com/valib/gopherdiscovery.git v0.0.0-20190531081358-bde1bf71186f/go.mod h1:CQ+UJyZV8MRzwwckncdUDu6/RDTKAzSIPCxc9tFcwPs=
github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo=
github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d h1:zE9ykElWQ6/NYmHa3jpm/yHnI4xSofP+UP6SpjHcSeM=
github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
github.com/smartystreets/goconvey v0.0.0-20190330032615-68dc04aab96a h1:pa8hGb/2YqsZKovtsgrwcDH1RZhVbTKCjLp47XpqCDs=
github.com/smartystreets/goconvey v0.0.0-20190330032615-68dc04aab96a/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
github.com/tmthrgd/go-sem v0.0.0-20160607101025-0214dbf53877/go.mod h1:sgTk9wg3WurMlziuB3hcfgHYTz3pEkjQpSCTT8V2pW8=
github.com/tmthrgd/go-shm v0.0.0-20170117044846-90afcfcd5ee9/go.mod h1:vy1jksyhzuQOMkHXMEi+X2bZ47ZeCn3QTnYdFBesABs=

 gopherdiscovery

New file
@@ -1 +1 @@
Subproject commit 0000000000000000000000000000000000000000
Subproject commit 3a3761054bec493abbc66c75aada07c669f28c8f

 main.go

@@ -5,7 +5,6 @@
    "log"
    "net/http"
    _ "net/http/pprof"
    "time"

    "github.com/long/test/camera"
    "github.com/long/test/sdk"
@@ -35,7 +34,5 @@
    sdk.Init()     //  获取所有算法id ,建立 sdk 主题， 建立sdk server(send, recv 运行)
    tasktag.Init() // 获取所有任务，建立任务标签， 在数据进入时， 打标签
    camera.Init()  //获取cid, taskid, sdkid ,关系
    for {
        time.Sleep(time.Second)
    }
    select {}
}

 sdk/sdk.go

@@ -21,6 +21,7 @@
)

var SocketManage = make(map[string]SocketContext)
var SdkMap = make(map[string]chan *protomsg.SdkMessage)

type SocketContext struct {
    Sock    deliver.Deliver
@@ -31,25 +32,48 @@
func Init() {

    fmt.Println("============= init sdk info =====================")
    SdkCreateTopic(util.Sdklist) // 创建主题

    for _, sdkid := range util.Sdklist { // 创建sdk server
        url := fmt.Sprintf("ipc:///tmp/%s%s", sdkid, postPull)
        sdkidser, socketser, err := NewSdkSocketListen(deliver.PushPull, sdkid, url)
        if err != nil {
            continue
        }
        go Send(sdkidser, socketser, SdkMap[sdkid])

        url = fmt.Sprintf("ipc:///tmp/%s%s", sdkid, postPush)
        _, socketdial, err := NewSdkSocketListen(deliver.PushPull, sdkid, url)
        if err != nil {
            continue
        }
        go Recv(socketdial)

        CreatesdkTopicandServer(sdkid)
        fmt.Println()
    }

    SdkMap["es"] = make(chan *protomsg.SdkMessage)
    fmt.Println("create es channel:  ")
    go es(SdkMap["es"])
    go AutoDelSdk(util.Sdkflag)
}

func CreatesdkTopicandServer(sdkid string) {
    SdkMap[sdkid] = make(chan *protomsg.SdkMessage)
    fmt.Println("create sdk channel:  ", sdkid)

    url := fmt.Sprintf("ipc:///tmp/%s%s", sdkid, postPull)
    sdkidser, socketser, err := NewSdkSocketListen(deliver.PushPull, sdkid, url)
    if err != nil {
        delete(SdkMap, sdkid)
        fmt.Println(sdkid, "create server error!")
        return
    }
    go Send(sdkidser, socketser, SdkMap[sdkid])

    url = fmt.Sprintf("ipc:///tmp/%s%s", sdkid, postPush)
    _, socketdial, err := NewSdkSocketListen(deliver.PushPull, sdkid, url)
    if err != nil {
        delete(SdkMap, sdkid)
        fmt.Println(sdkid, "create dial error!")
        return
    }
    go Recv(socketdial)

}
func DeletesdkTopicandServer(sdkid string) {
    close(SdkMap[sdkid])
    delete(SdkMap, sdkid)
    fmt.Println("删除主题 sdk: ", sdkid)

    SocketManage[sdkid].Cancel()
    delete(SocketManage, sdkid)
    fmt.Println("删除server sdk: ", sdkid)
}

//单独处理   es 主题的情况
@@ -60,38 +84,50 @@
}

//动态处理
func AutoDelSdk(Newsdklist []string) {
    var oldSdk []string
    for key, _ := range SdkMap {
        oldSdk = append(oldSdk, key)
    }
    sdkChanDel := util.Difference(oldSdk, Newsdklist)
func AutoDelSdk(sdkflag chan bool) {

    for key, op := range sdkChanDel {
        if op == "add" {
            SdkMap[key] = make(chan *protomsg.SdkMessage)
            fmt.Println("创建主题 sdk: ", key)
        } else {
            close(SdkMap[key])
            delete(SdkMap, key)
            fmt.Println("删除主题 sdk: ", key)
    for _ = range sdkflag {
        fmt.Println("test autodelsdk")
        var oldSdk []string
        for key, _ := range SdkMap {
            oldSdk = append(oldSdk, key)
        }
        util.Sdklist = append(util.Sdklist, "es")
        sdkChanDel := util.Difference(oldSdk, util.Sdklist)
        fmt.Println(sdkChanDel)

        for key, op := range sdkChanDel {
            if op == "add" {
                CreatesdkTopicandServer(key)
            } else {
                DeletesdkTopicandServer(key)
                fmt.Println("删除主题 sdk: ", key)
            }
        }
    }
}

//主题
var SdkMap = make(map[string]chan *protomsg.SdkMessage)

//sdk数据 加工器
func SdkData(cid string, taskid string, data []byte) *protomsg.SdkMessage {
    var sdkmsg = &protomsg.SdkMessage{}
    sdkmsg.Cid = cid
    if _, ok := tasktag.TaskMapLab[taskid]; !ok {
    //if _, ok := tasktag.TaskMapLab[taskid]; !ok {
    //    sdkmsg.Tasklab = nil
    //    return sdkmsg
    //}

    if val, ok := tasktag.TaskMapLab.Load(taskid); ok {
        sdkmsg.Tasklab = nil
        return sdkmsg

    } else {

        sdkmsg.Tasklab = val.(*protomsg.TaskLabel)
        sdkmsg.Data = data
    }
    sdkmsg.Tasklab = tasktag.TaskMapLab[taskid]
    sdkmsg.Data = data
    //sdkmsg.Tasklab = tasktag.TaskMapLab[taskid]
    return sdkmsg
}

@@ -106,19 +142,6 @@
    return
}

// 创建主题
func SdkCreateTopic(sdklist []string) (err error) {
    for _, sdkid := range sdklist {
        SdkMap[sdkid] = make(chan *protomsg.SdkMessage)
        fmt.Println("create sdk channel:  ", sdkid)
    }

    SdkMap["es"] = make(chan *protomsg.SdkMessage)
    fmt.Println("create es channel:  ")

    return nil
}

// create server
func NewSdkSocketListen(mode int, sdkid string, url string) (sid string, socket SocketContext, err error) {
    fmt.Println("url is: ", url)
@@ -128,7 +151,6 @@
    socket.Cancel = cancel

    socket.Sock = deliver.NewServer(deliver.Mode(mode), url)
    fmt.Println(sdkid, socket.Sock)

    if socket.Sock == nil {
        return sdkid, socket, errors.New("create listen error")
@@ -146,7 +168,6 @@
    socket.Cancel = cancel

    socket.Sock = deliver.NewClient(deliver.Mode(mode), url)
    fmt.Println(sdkid, socket.Sock)

    if socket.Sock == nil {
        return sdkid, socket, errors.New("create listen error")
@@ -157,7 +178,6 @@
}

func Recv(socket SocketContext) {
    //socket.Sock.SetOption(mangos.OptionRecvDeadline, 1*time.Second)

    var repsdkmsg = &protomsg.SdkMessage{}
    for {
@@ -186,27 +206,35 @@

func Send(sdkid string, socket SocketContext, in chan *protomsg.SdkMessage) {
    var v *protomsg.SdkMessage
    var ok bool

    for {
        select {
        case <-socket.Context.Done():
            fmt.Println("socket is close")
        case v = <-in:
            data, err := proto.Marshal(v)
            if err != nil {
                fmt.Println("proto marshal error ", err)
                continue
            }
            fmt.Printf("从管道sdkid=%s 接受数据 %d\n", sdkid, len(data))
            fmt.Println()
            return
        case v, ok = <-in:
            if ok {
                data, err := proto.Marshal(v)
                if err != nil {
                    fmt.Println("proto marshal error ", err)
                    continue
                }
                fmt.Printf("从管道sdkid=%s 接受数据 %d\n", sdkid, len(data))
                fmt.Println()

            fmt.Println("send len of data: ", len(data))
            if err := socket.Sock.Send(data); err != nil {
                fmt.Println(socket.Sock)
                fmt.Println("failed send")
                continue
                fmt.Println("send len of data: ", len(data))
                if err := socket.Sock.Send(data); err != nil {
                    fmt.Println(socket.Sock)
                    fmt.Println("failed send")
                    continue
                }
                fmt.Printf("sdkid = %s ,send success:%d \n", sdkid, len(data))
            } else {
                fmt.Println(sdkid, " 主题关闭, 关闭send()")
                return
            }
            fmt.Printf("sdkid = %s ,send success:%d \n", sdkid, len(data))

        }
    }
}

 tasktag/tasktag.go

@@ -2,18 +2,30 @@

import (
    "fmt"
    "sync"

    "basic.com/pubsub/protomsg.git"
    "github.com/long/test/util"
)

var TaskMapLab = make(map[string]*protomsg.TaskLabel)
//var TaskMapLab = make(map[string]*protomsg.TaskLabel)
var TaskMapLab sync.Map

func Init() {

    var tls []protomsg.TaskLabel
    fmt.Println("============= init tasktag info =====================")
    GenTaskMap()
    go func(taskflag chan bool) {
        for _ = range taskflag {
            GenTaskMap()
            fmt.Println("update task finished!")
        }

    }(util.TaskSdkflag)
}

func GenTaskMap() {
    var tls []protomsg.TaskLabel
    for _, taskSdk := range util.TaskSdks {
        var tl protomsg.TaskLabel
        tl.Taskid = taskSdk.Task.Taskid
@@ -26,17 +38,30 @@

    GenTasklab(tls)

    for key, value := range TaskMapLab {
        fmt.Println()
        fmt.Println(key, value)
    }
    TaskMapLab.Range(func(k, v interface{}) bool {
        fmt.Println(k, v)
        return true
    })

    //for key, value := range TaskMapLab {
    //    fmt.Println()
    //    fmt.Println(key, value)
    //}
}

// 从sqlite 接口拿到所有的任务, 每一个任务都有自己的几个算法
//以 taskid 作为key, 对应的算法组合作为 value
func GenTasklab(tasklab []protomsg.TaskLabel) {
    // TaskMapLab = nil
    // TaskMapLab = make(map[string]*protomsg.TaskLabel)
    TaskMapLab.Range(func(key interface{}, value interface{}) bool {
        TaskMapLab.Delete(key)
        return true
    })

    for _, value := range tasklab {
        pv := value
        TaskMapLab[value.Taskid] = &pv
        TaskMapLab.Store(value.Taskid, &pv)
        //TaskMapLab[value.Taskid] = &pv
    }
}

 test

Binary files differ

 util/sqlite.go

@@ -5,8 +5,8 @@

    "basic.com/pubsub/protomsg.git"
    "basic.com/r/dbapi.git"
    "basic.com/valib/gopherdiscovery.git"
    "github.com/gogo/protobuf/proto"
    "github.com/long/test/gopherdiscovery"
)

/*************************
@@ -26,54 +26,74 @@
var TaskSdks []protomsg.TaskSdkInfo
var Sdklist []string

var urlServer = "tcp://192.168.1.11:40007"
var urlPubSub = "tcp://192.168.1.11:50007"
var peers chan []string
//var urlServer = "tcp://192.168.1.11:40007"
//var urlPubSub = "tcp://192.168.1.11:50007"

var urlServer = "tcp://127.0.0.1:40007"
var urlPubSub = "tcp://127.0.0.1:50007"

var Sdkflag = make(chan bool)
var TaskSdkflag = make(chan bool)
var Cameraflag = make(chan bool)

var newsdkmsg = &protomsg.DbChangeMessage{}

func Getdata(opt []byte, initchan chan bool) {
func processinit(initchan chan bool) {
    CameraIds = camval.FindAll()

    fmt.Println("==============camera camera with task ================")
    CameraTasks = camval.FindAllCameraAndTask()
    fmt.Println(CameraTasks)

    TaskSdks = taskapi.FindAll()

    Sdklist = sdkapi.GetAllSdkIds()
    initchan <- true
}

func Getdata(opt []byte) {
    if err := proto.Unmarshal(opt, newsdkmsg); err != nil {
        fmt.Println("publichshMessage ", err)
        return
    }

    if !IsInit {
        fmt.Println("enter init.")
        CameraIds = camval.FindAll()

        fmt.Println("==============camera camera with task ================")
        CameraTasks = camval.FindAllCameraAndTask()
        fmt.Println(CameraTasks)

        TaskSdks = taskapi.FindAll()

        Sdklist = sdkapi.GetAllSdkIds()
        fmt.Println(Sdklist)
        IsInit = true
        initchan <- true
        return
    }

    switch newsdkmsg.Table {
    case protomsg.TableChanged_T_Camera:
        fmt.Println("update camera")
        CameraIds = camval.FindAll()
    case protomsg.TableChanged_T_CameraTask:
        CameraTasks = camval.FindAllCameraAndTask()
    case protomsg.TableChanged_T_TaskSdk:
        TaskSdks = taskapi.FindAll()
    case protomsg.TableChanged_T_Sdk:
        Sdklist = sdkapi.GetAllSdkIds()
    }
        Cameraflag <- true

    case protomsg.TableChanged_T_CameraTask:
        fmt.Println("update cameratask")
        CameraTasks = camval.FindAllCameraAndTask()
        fmt.Println("update cameratask finished!")

    case protomsg.TableChanged_T_TaskSdk:
        fmt.Println("update tasksdk")
        TaskSdks = taskapi.FindAll()
        TaskSdkflag <- true

    case protomsg.TableChanged_T_Sdk:
        fmt.Println("update sdk")
        Sdklist = sdkapi.GetAllSdkIds()
        Sdkflag <- true
        fmt.Println("update finished!")

    default:
        fmt.Println("unknow type operation")
    }
    fmt.Println(newsdkmsg)
}

func Init(initchan chan bool) {
    clientOne, _ := gopherdiscovery.ClientWithSub(urlServer, urlPubSub, "ip:192.168.1.124")
    peers, _ = clientOne.Peers()
    recvinit := clientOne.HeartBeatMsg()
    fmt.Println(<-recvinit)
    processinit(initchan)

    peers, _ := clientOne.Peers()
    for x := range peers {
        Getdata([]byte(x[0]), initchan)
        fmt.Println("client: ", x)
        Getdata(x)
    }
}