add heartbeat

龙赣华

2019-05-31 3c8609a3da72cb2f3610f3f1789dafacc6c0f40f

add heartbeat

 camera/camera.go

@@ -1,141 +1,138 @@
package camera

import (
        "errors"
    "errors"

        "basic.com/dbapi.git"
        "basic.com/valib/deliver.git"
        "basic.com/pubsub/protomsg.git"
    "basic.com/dbapi.git"
    "basic.com/valib/deliver.git"

        "github.com/long/test/sdk"
    "github.com/long/test/sdk"
    "github.com/long/test/util"

        "context"
        "fmt"
        "sync"
        "time"
       )
    "context"
    "fmt"
    "sync"
    "time"
)

//var SocketManage = make(map[string]SocketContext)
var SocketManage sync.Map

var Initchannel = make(chan string)

    type SocketContext struct {
        Sock    deliver.Deliver
            Context context.Context
            Cancel  context.CancelFunc
    }
type SocketContext struct {
    Sock    deliver.Deliver
    Context context.Context
    Cancel  context.CancelFunc
}

var camval dbapi.CameraApi

//get camera with task
var cameraTasks []protomsg.CameraAndTaskInfo

func Init() {

    fmt.Println("============ camera info ====================")
    ctsid := camval.FindAll()
    fmt.Println("============ camera info ====================")
    for _, cd := range util.CameraIds {
        fmt.Println(cd)
        fmt.Println()
    }

    fmt.Println("==============camera camera with task ================")
    cameraTasks=camval.FindAllCameraAndTask()
    go CreateCamera(Initchannel)

    go CreateCamera(Initchannel)

    for _, cam := range ctsid {
        Initchannel <- cam.Id
    }
    for _, cam := range util.CameraIds {
        Initchannel <- cam.Id
    }
}

func CreateCamera(camera chan string) {
    for camid := range camera {
        if _, ok := SocketManage.Load(camid); !ok {
            url := fmt.Sprintf("ipc:///tmp/%s.ipc", camid)
    for camid := range camera {
        if _, ok := SocketManage.Load(camid); !ok {
            url := fmt.Sprintf("ipc:///tmp/%s.ipc", camid)

            id, socketlisten, err := NewCamerSocketListen(deliver.PushPull, camid, url)
            if err != nil {
              fmt.Println("create socket error")
              continue 
            }
            id, socketlisten, err := NewCamerSocketListen(deliver.PushPull, camid, url)
            if err != nil {
                fmt.Println("create socket error")
                continue
            }

             go func(cid string, sock SocketContext ){
                  Recv(cid,sock)
             }(id,socketlisten)
        }
    }
            go func(cid string, sock SocketContext) {
                Recv(cid, sock)
            }(id, socketlisten)
        }
    }
}

// create server
func NewCamerSocketListen(mode int, cameraid string, url string) (cid string, socket SocketContext, err error) {
        ctx, cancel := context.WithCancel(context.Background())
    ctx, cancel := context.WithCancel(context.Background())

        socket.Context = ctx
        socket.Cancel = cancel
    socket.Context = ctx
    socket.Cancel = cancel

        socket.Sock = deliver.NewServer(deliver.Mode(mode), url)
        fmt.Println("new socket.Sock: ", socket.Sock)
    socket.Sock = deliver.NewServer(deliver.Mode(mode), url)
    fmt.Println("new socket.Sock: ", socket.Sock)

        if socket.Sock == nil {
            return cameraid, socket, errors.New("create listen error")
        }
        SocketManage.Store(cameraid, socket)
        return cameraid, socket, nil
    if socket.Sock == nil {
        return cameraid, socket, errors.New("create listen error")
    }
    SocketManage.Store(cameraid, socket)
    return cameraid, socket, nil
}

func Recv(cameraid string, socket SocketContext) {
        var msg []byte
        var err error
        for {
            select {
                case <-socket.Context.Done():
                    fmt.Println("listen recv quit")
                        return
                default:
                        if msg, err = socket.Sock.Recv(); err != nil {
                            fmt.Println("err is: ", cameraid, err)
                                continue
                        } else {
                            fmt.Println()
                            fmt.Println("============== one msg input ==========")
                            fmt.Println("cameraid: ",cameraid,  len(msg))
                            for _, taskid := range GetAlltask(cameraid) {
                                time.Sleep(5* time.Second)
                                fmt.Println("cameraid: ",cameraid," taskid: ", taskid)
                                Taskdolist(cameraid, taskid, msg)
                             }
                        }
            }
        }
    var msg []byte
    var err error
    for {
        select {
        case <-socket.Context.Done():
            fmt.Println("listen recv quit")
            return
        default:
            if msg, err = socket.Sock.Recv(); err != nil {
                fmt.Println("err is: ", cameraid, err)
                continue
            } else {
                fmt.Println()
                fmt.Println("============== one msg input ==========")
                fmt.Println("cameraid: ", cameraid, len(msg))
                for _, taskid := range GetAlltask(cameraid) {
                    time.Sleep(5 * time.Second)
                    fmt.Println("cameraid: ", cameraid, " taskid: ", taskid)
                    Taskdolist(cameraid, taskid, msg)
                }
            }
        }
    }
}

//   根据cid 获取 所有的任务
func GetAlltask(cid string) (tasks []string) {
    for _, camsingle := range cameraTasks {
        if cid == camsingle.Camera.Id {
            for _, tasksingle := range camsingle.Tasks {
                tasks = append(tasks, tasksingle.Taskid)
            }
                return
        }
    }
    return
    for _, camsingle := range util.CameraTasks {
        if cid == camsingle.Camera.Id {
            for _, tasksingle := range camsingle.Tasks {
                tasks = append(tasks, tasksingle.Taskid)
            }
            return
        }
    }
    return
}

func Taskdolist(cid string, taskid string, data []byte) {

        //  数据加工(打标签)
        sdkmsg := sdk.SdkData(cid, taskid, data)
        if sdkmsg.Tasklab == nil {
            fmt.Println("cid:%s 没有任务%s", cid, taskid)
                return
        }
    //  数据加工(打标签)
    sdkmsg := sdk.SdkData(cid, taskid, data)
    if sdkmsg.Tasklab == nil {
        fmt.Println("cid:%s 没有任务%s", cid, taskid)
        return
    }

    //  计算分发的主题
            SendTopic := sdk.SdkSendTopic(sdkmsg)
               if _, ok := sdk.SdkMap[SendTopic]; ok {
                   fmt.Println("分发的主题存在")
                       sdk.SdkMap[SendTopic] <- sdkmsg
                       //fmt.Println("重新开始循环: ", sdk.SdkMap)
               } else {
                   fmt.Println("分发的主题不存在")
               }
    //  计算分发的主题
    SendTopic := sdk.SdkSendTopic(sdkmsg)
    if _, ok := sdk.SdkMap[SendTopic]; ok {
        fmt.Println("分发的主题存在")
        sdk.SdkMap[SendTopic] <- sdkmsg
        //fmt.Println("重新开始循环: ", sdk.SdkMap)
    } else {
        fmt.Println("分发的主题不存在")
    }
}

 go.mod

@@ -4,14 +4,17 @@

require (
    basic.com/dbapi.git v0.0.0-20190523103034-c0d33072098b
    basic.com/pubsub/protomsg.git v0.0.0-20190524072049-ce8c0f2b14dd
    basic.com/valib/deliver.git v0.0.0-20190522053509-c2bbe31e7c6c
    github.com/gogo/protobuf v1.2.1 // indirect
    basic.com/pubsub/protomsg.git v0.0.0-20190530084829-0ea842491a96
    basic.com/r/dbapi.git v0.0.0-20190523103034-c0d33072098b
    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/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
    golang.org/x/sys v0.0.0-20190522044717-8097e1b27ff5 // indirect
    nanomsg.org/go-mangos v1.4.0
    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
)

 go.sum

@@ -1,31 +1,32 @@
basic.com/dbapi.git v0.0.0-20190523025708-eaf1da6b55de h1:tkAqiVXaBz8upBGGu60jExv0H5H7m2OWZdR8aTAJkp0=
basic.com/dbapi.git v0.0.0-20190523025708-eaf1da6b55de/go.mod h1:eDXPnxaz6jZPDvBSk7ya7oSASWPCuUEgRTJCjsfKt/Q=
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/pubsub/protomsg.git v0.0.0-20190523080134-c2459cf7ffa7 h1:AmdkBGk95CJy0jMU2SJkQNgldpPCMyAifeAPxUngQsw=
basic.com/pubsub/protomsg.git v0.0.0-20190523080134-c2459cf7ffa7/go.mod h1:un5NV5VWQoblVLZfx1Rt5vyLgwR0jI92d3VJhfrJhWU=
basic.com/pubsub/protomsg.git v0.0.0-20190524044418-e6c6e5fdcdab h1:kTHZgvhdEJ+Vdbi1bBhKRA2oTYMhLZDqpWWk40yUd3s=
basic.com/pubsub/protomsg.git v0.0.0-20190524044418-e6c6e5fdcdab/go.mod h1:un5NV5VWQoblVLZfx1Rt5vyLgwR0jI92d3VJhfrJhWU=
basic.com/pubsub/protomsg.git v0.0.0-20190524072049-ce8c0f2b14dd h1:Z1KVegr3JrNHaJFAv6yRniNIWdvzLWBPkpBRnpzgnYg=
basic.com/pubsub/protomsg.git v0.0.0-20190524072049-ce8c0f2b14dd/go.mod h1:un5NV5VWQoblVLZfx1Rt5vyLgwR0jI92d3VJhfrJhWU=
basic.com/valib/deliver.git v0.0.0-20190522053509-c2bbe31e7c6c h1:nyclQo40lBhvt2LnsaG/tNyxuotKou0V67jL1jBcJfM=
basic.com/valib/deliver.git v0.0.0-20190522053509-c2bbe31e7c6c/go.mod h1:bkYiTUGzckyNOjAgn9rB/DOjFzwoSHJlruuWQ6hu6IY=
github.com/gogo/protobuf v1.2.1 h1:/s5zKNz0uPFCZ5hddgPdo2TK2TVrUNMn0OOX8/aZMTE=
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/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=
github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
github.com/golang/protobuf v1.3.1 h1:YF8+flBXS5eO826T4nzqPrxfhQThhXl0YzfuUPu4SBg=
github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
github.com/gorilla/websocket v1.4.0 h1:WDFjx/TMzVgy9VdMMQi2K2Emtwi2QcUQsztZ/zLaH/Q=
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/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/tmthrgd/go-sem v0.0.0-20160607101025-0214dbf53877 h1:n65+IT/xy5+trHm3Zpg9+j7IO4n8pBcPzvaKbMolW8U=
github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
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 h1:uVRQSWD6TOlWlLJ7IYYmbjRr0Xg35ADFN89HGQLPFGI=
github.com/tmthrgd/go-shm v0.0.0-20170117044846-90afcfcd5ee9/go.mod h1:vy1jksyhzuQOMkHXMEi+X2bZ47ZeCn3QTnYdFBesABs=
github.com/tmthrgd/shm-go v0.0.0-20170130075737-7207ca97b290 h1:5zW+TRr0WH4uN72/E/XYwb1PcaYN5BIB/FUbcQ0nHr0=
github.com/tmthrgd/shm-go v0.0.0-20170130075737-7207ca97b290/go.mod h1:e9PZQr6zVezMTwj1v0j1YhGCNdS2zTCjXU9q9K+HHGk=
golang.org/x/sys v0.0.0-20190522044717-8097e1b27ff5 h1:f005F/Jl5JLP036x7QIvUVhNTqxvSYwFIiyOh2q12iU=
golang.org/x/sys v0.0.0-20190522044717-8097e1b27ff5/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/net v0.0.0-20190522155817-f3200d17e092 h1:4QSRKanuywn15aTZvI/mIDEgPQpswuFndXpOj3rKEco=
golang.org/x/net v0.0.0-20190522155817-f3200d17e092/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a h1:1BGLXjeY4akVXGgbC9HugT3Jv3hCI0z56oJR5vAMgBU=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/tools v0.0.0-20180221164845-07fd8470d635/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
nanomsg.org/go-mangos v1.4.0 h1:pVRLnzXePdSbhWlWdSncYszTagERhMG5zK/vXYmbEdM=
golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
nanomsg.org/go-mangos v1.4.0/go.mod h1:MOor8xUIgwsRMPpLr9xQxe7bT7rciibScOqVyztNxHQ=

 httpclient/es_test.go

File was deleted

 httpclient/esutil.go

File was deleted

 main.go

@@ -1,24 +1,37 @@
package main

import (
    "fmt"
    "log"
    "net/http"
    _ "net/http/pprof"
    "time"

    "github.com/long/test/camera"

    //    "github.com/long/test/httpclient"
    "github.com/long/test/sdk"
    "github.com/long/test/tasktag"
    "time"
    "github.com/long/test/util"
    //    "github.com/long/test/camera"
    //    "github.com/long/test/sdk"
    //    "github.com/long/test/tasktag"
)

var initchan = make(chan bool)

func main() {

    // pprof 用于分析性能
    go func() {
        log.Println(http.ListenAndServe("192.168.1.123:6060", nil))
        log.Println(http.ListenAndServe("0.0.0.0:6060", nil))
    }()

    go util.Init(initchan)

    fmt.Println("init ok !!!!, start sdk, task, camera init process ....", <-initchan)
    fmt.Println()
    fmt.Println()
    fmt.Println()
    fmt.Println()
    sdk.Init()     //  获取所有算法id ,建立 sdk 主题， 建立sdk server(send, recv 运行)
    tasktag.Init() // 获取所有任务，建立任务标签， 在数据进入时， 打标签
    camera.Init()  //获取cid, taskid, sdkid ,关系

 sdk/sdk.go

@@ -5,25 +5,22 @@
    "errors"
    "fmt"

//    "github.com/long/test/httpclient"
    //    "github.com/long/test/httpclient"
    "github.com/long/test/tasktag"
    "github.com/long/test/util"

    "github.com/golang/protobuf/proto"

    "basic.com/valib/deliver.git"
    "basic.com/pubsub/protomsg.git"
    "basic.com/dbapi.git"
    "basic.com/valib/deliver.git"
)

const (
    postPull="_1.ipc"
    postPush="_2.ipc"
    postPull = "_1.ipc"
    postPush = "_2.ipc"
)

var SocketManage = make(map[string]SocketContext)

var sdkapi dbapi.SdkApi

type SocketContext struct {
    Sock    deliver.Deliver
@@ -33,19 +30,18 @@

func Init() {

    fmt.Println("============= init sdk info =====================")
    sdklist := SdkAll() //获取所有sdk
    SdkCreateTopic(sdklist) // 创建主题
    fmt.Println("============= init sdk info =====================")
    SdkCreateTopic(util.Sdklist) // 创建主题

    for _, sdkid := range sdklist { // 创建sdk server
        url := fmt.Sprintf("ipc:///tmp/%s%s",sdkid,postPull)
    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)
        url = fmt.Sprintf("ipc:///tmp/%s%s", sdkid, postPush)
        _, socketdial, err := NewSdkSocketListen(deliver.PushPull, sdkid, url)
        if err != nil {
            continue
@@ -110,13 +106,6 @@
    return
}

// 调用  http 借口获取摄像机信息
func SdkAll() (sdklist []string) {
    //sdklist = httpclient.GetSdk("http://192.168.1.124:8000/data/api-v/sdk/findskdid")
    sdklist = sdkapi.GetAllSdkIds()
    return
}

// 创建主题
func SdkCreateTopic(sdklist []string) (err error) {
    for _, sdkid := range sdklist {
@@ -129,7 +118,6 @@

    return nil
}


// create server
func NewSdkSocketListen(mode int, sdkid string, url string) (sid string, socket SocketContext, err error) {
@@ -214,7 +202,7 @@

            fmt.Println("send len of data: ", len(data))
            if err := socket.Sock.Send(data); err != nil {
                fmt.Println(socket.Sock)
                fmt.Println(socket.Sock)
                fmt.Println("failed send")
                continue
            }

 tasktag/tasktag.go

@@ -2,30 +2,27 @@

import (
    "fmt"

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

var TaskMapLab = make(map[string] *protomsg.TaskLabel)
var taskapi dbapi.TaskApi
var TaskMapLab = make(map[string]*protomsg.TaskLabel)

func Init() {

    fmt.Println("=========== tasktag info ====================")
    var tls []protomsg.TaskLabel
    fmt.Println("============= init tasktag info =====================")

     taskSdks := taskapi.FindAll()
     for _, taskSdk := range taskSdks {
         fmt.Println("test:    ", taskSdk)
         var tl  protomsg.TaskLabel
         tl.Taskid = taskSdk.Task.Taskid
         for _, sdkinfo := range taskSdk.Sdks {
             tl.Sdkids = append(tl.Sdkids, sdkinfo.Id)
         }
         fmt.Println(tl)
         tl.Index=int32(0)
         tls = append(tls, tl)
     }
    for _, taskSdk := range util.TaskSdks {
        var tl protomsg.TaskLabel
        tl.Taskid = taskSdk.Task.Taskid
        for _, sdkinfo := range taskSdk.Sdks {
            tl.Sdkids = append(tl.Sdkids, sdkinfo.Id)
        }
        tl.Index = int32(0)
        tls = append(tls, tl)
    }

    GenTasklab(tls)

@@ -39,7 +36,7 @@
//以 taskid 作为key, 对应的算法组合作为 value
func GenTasklab(tasklab []protomsg.TaskLabel) {
    for _, value := range tasklab {
        pv := value
        TaskMapLab[value.Taskid] = &pv  
        pv := value
        TaskMapLab[value.Taskid] = &pv
    }
}

 test

Binary files differ

 util/sqlite.go

New file
@@ -0,0 +1,79 @@
package util

import (
    "fmt"

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

/*************************
* 1. store all struct data
* 2. define function for get difference data type
* 3. recive data from sqlite process. and call conpos function.
***********************/

var camval dbapi.CameraApi
var taskapi dbapi.TaskApi
var sdkapi dbapi.SdkApi
var IsInit bool

var CameraIds []protomsg.Camera
var CameraTasks []protomsg.CameraAndTaskInfo

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 newsdkmsg = &protomsg.DbChangeMessage{}

func Getdata(opt []byte, initchan chan bool) {

    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:
        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()
    }

    fmt.Println(newsdkmsg)
}

func Init(initchan chan bool) {
    clientOne, _ := gopherdiscovery.ClientWithSub(urlServer, urlPubSub, "ip:192.168.1.124")
    peers, _ = clientOne.Peers()
    for x := range peers {
        Getdata([]byte(x[0]), initchan)
    }
}