From f963cd025c1aa88ac8b211e24f46ceb0eb64c418 Mon Sep 17 00:00:00 2001 From: zhangmeng <775834166@qq.com> Date: 星期二, 07 五月 2019 14:19:29 +0800 Subject: [PATCH] 多个解码单元同时运行,注意tcp地址 --- decoder/main.go | 95 ++++++++++++++++++++++++++++++----------------- 1 files changed, 61 insertions(+), 34 deletions(-) diff --git a/decoder/main.go b/decoder/main.go index c6e75b8..4684779 100644 --- a/decoder/main.go +++ b/decoder/main.go @@ -2,12 +2,13 @@ import ( "context" - "decoder/demo" "decoder/valib/ipc" + srv "decoder/work/service" "encoding/json" "flag" "fmt" - // "videoServer/demo" + "strconv" + "strings" ) var ( @@ -16,6 +17,10 @@ ipcURL string proc string + + testIt bool + + asServer bool ) func init() { @@ -23,6 +28,10 @@ flag.StringVar(&picFolder, "f", ".", "test pic folder") flag.StringVar(&ipcURL, "ipc", "ipc:///tmp/pic.ipc", "ipc label") + + flag.BoolVar(&testIt, "test", false, "use test") + + flag.BoolVar(&asServer, "server", false, "run ipc as server") } // CameraInfo camera info @@ -38,28 +47,36 @@ } var ( - mapCI = make(map[string]CameraInfo) - port = 7001 + mapCameraInfo = make(map[string]CameraInfo) + + tcp = `tcp://192.168.1.124:` + port = 7001 ) -func recvFromIPC(ctx context.Context, url string, ch chan<- CameraInfo) { +func recvCameraInfoFromIPC(ctx context.Context, url string, ch chan<- CameraInfo) { ipc := ipc.NewClient(ctx, url) for { msg := ipc.Recv() if msg != nil { - fmt.Println(string(msg)) + var c CameraInfo if err := json.Unmarshal(msg, &c); err == nil { + + if _, ok := mapCameraInfo[c.ID]; ok { + continue + } + ch <- c - fmt.Printf("recv camere info %+v\n", c) + + msgIpc := MsgIPC{"new decoder", port} + if b, err := json.Marshal(msgIpc); err == nil { + ipc.Send(b) + } } else { fmt.Println(err) } - msgIpc := MsgIPC{"new decoder", port} - if b, err := json.Marshal(msgIpc); err == nil { - ipc.Send(b) - } + } } } @@ -67,33 +84,43 @@ func main() { flag.Parse() - test() + if testIt { + test() + } - // ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithCancel(context.Background()) - // ch := make(chan CameraInfo) - // go recvFromIPC(ctx, "tcp://192.168.1.124:7000", ch) + ch := make(chan CameraInfo) + // tcpURL := tcp + strconv.Itoa(port) + // port++ + go recvCameraInfoFromIPC(ctx, ipcURL, ch) - // for { - // select { - // case <-ctx.Done(): - // return - // case c := <-ch: - // if _, ok := mapCI[c.ID]; !ok { - // mapCI[c.ID] = c - // ipc := "tcp://192.168.1.124:" + strconv.Itoa(port) - // port++ - // fmt.Printf("create ipc %s for decode : %s, on camera id %s\n", ipc, c.URL, c.ID) + for { + select { + case <-ctx.Done(): + return + case c := <-ch: + if _, ok := mapCameraInfo[c.ID]; !ok { + mapCameraInfo[c.ID] = c + ipcAddr := tcp + strconv.Itoa(port) + port++ - // url := strings.TrimSpace(c.URL) - // id := strings.TrimSpace(c.ID) - // i := strings.TrimSpace(ipc) - // go demo.SendByIPC(url, id, i) - // } - // } - // } - // cancel() + url := strings.TrimSpace(c.URL) + id := strings.TrimSpace(c.ID) + addr := strings.TrimSpace(ipcAddr) + go runSender(id, url, addr) + } + } + } + cancel() +} +func runSender(cameraID, rtspURL, ipcLabel string) { + d := srv.NewSender(cameraID, rtspURL, ipcLabel) + if asServer { + d.RunAsServer() + } + d.RunAsClient() } func test() { @@ -101,5 +128,5 @@ fmt.Println(picFolder) - demo.SendByIPC(streamURL, "camera1", ipcURL, true) + runSender("cameraid", streamURL, ipcURL) } -- Gitblit v1.8.0