From d3618c4e9ceab273a52813c1f2c462912fb81e59 Mon Sep 17 00:00:00 2001 From: zhangmeng <775834166@qq.com> Date: 星期二, 30 四月 2019 15:40:14 +0800 Subject: [PATCH] add ipc run as server or not --- decoder/work/service/reciever.go | 11 ++++- decoder/main.go | 88 ++++++++++++++++++++++++++++++------------- decoder/valib/ipc | 2 decoder/work/service/decoder.go | 13 ++++-- decoder/demo/simpleIPC.go | 4 +- 5 files changed, 81 insertions(+), 37 deletions(-) diff --git a/decoder/demo/simpleIPC.go b/decoder/demo/simpleIPC.go index 8f309f0..aee52c8 100644 --- a/decoder/demo/simpleIPC.go +++ b/decoder/demo/simpleIPC.go @@ -5,7 +5,7 @@ ) // SendByIPC send pic by ipc -func SendByIPC(rtspURL, cameraID, ipcURL string) { +func SendByIPC(rtspURL, cameraID, ipcURL string, asServer bool) { d := srv.NewDecoder(cameraID, rtspURL, ipcURL) - d.Run() + d.Run(asServer) } diff --git a/decoder/main.go b/decoder/main.go index 0c7f65b..c6e75b8 100644 --- a/decoder/main.go +++ b/decoder/main.go @@ -1,11 +1,10 @@ package main import ( - "bytes" "context" "decoder/demo" "decoder/valib/ipc" - "encoding/gob" + "encoding/json" "flag" "fmt" // "videoServer/demo" @@ -26,46 +25,81 @@ flag.StringVar(&ipcURL, "ipc", "ipc:///tmp/pic.ipc", "ipc label") } -func test() { - fmt.Println("start test") - - fmt.Println(picFolder) - - demo.SendByIPC(streamURL, "camera1", ipcURL) +// CameraInfo camera info +type CameraInfo struct { + ID string `json:"Cameraid"` + URL string `json:"Rtsp"` } -type cameraInfo struct { - cameraID string - videoURL string +// MsgIPC msg for ipc +type MsgIPC struct { + CMD string `json:"Command"` + Port int `jsong:"PortIpc"` } -func recvFromIPC(ctx context.Context, url string) (cameraID, rtspURL string) { +var ( + mapCI = make(map[string]CameraInfo) + port = 7001 +) + +func recvFromIPC(ctx context.Context, url string, ch chan<- CameraInfo) { ipc := ipc.NewClient(ctx, url) for { msg := ipc.Recv() - - var buf bytes.Buffer - buf.Write(msg) - - dec := gob.NewDecoder(&buf) - - var i cameraInfo - if err := dec.Decode(&i); err != nil { - fmt.Println("gob decode CameraImage error", err) - continue + if msg != nil { + fmt.Println(string(msg)) + var c CameraInfo + if err := json.Unmarshal(msg, &c); err == nil { + ch <- c + fmt.Printf("recv camere info %+v\n", c) + } else { + fmt.Println(err) + } + msgIpc := MsgIPC{"new decoder", port} + if b, err := json.Marshal(msgIpc); err == nil { + ipc.Send(b) + } } - return i.cameraID, i.videoURL } } func main() { flag.Parse() - ctx, cancel := context.WithCancel(context.Background()) - recvFromIPC(ctx, "tcp://192.168.1.156:7000") + test() - cancel() - // test() + // ctx, cancel := context.WithCancel(context.Background()) + // ch := make(chan CameraInfo) + // go recvFromIPC(ctx, "tcp://192.168.1.124:7000", 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) + + // url := strings.TrimSpace(c.URL) + // id := strings.TrimSpace(c.ID) + // i := strings.TrimSpace(ipc) + // go demo.SendByIPC(url, id, i) + // } + // } + // } + // cancel() + +} + +func test() { + fmt.Println("start test") + + fmt.Println(picFolder) + + demo.SendByIPC(streamURL, "camera1", ipcURL, true) } diff --git a/decoder/valib/ipc b/decoder/valib/ipc index 66d8e66..8841f9d 160000 --- a/decoder/valib/ipc +++ b/decoder/valib/ipc @@ -1 +1 @@ -Subproject commit 66d8e6695ada4090bdb81279d594f323be875bc7 +Subproject commit 8841f9d2154a8e0653d30fdc04bde8a772665c8c diff --git a/decoder/work/service/decoder.go b/decoder/work/service/decoder.go index dfff34d..c6f6aeb 100644 --- a/decoder/work/service/decoder.go +++ b/decoder/work/service/decoder.go @@ -55,8 +55,13 @@ } // Run run a decoder -func (dec *Decoder) Run() { - ipc := ipc.NewServer(dec.ctxIPC.ctx, dec.ipcURL) +func (dec *Decoder) Run(asServer bool) { + var i *ipc.IPC + if asServer { + i = ipc.NewServer(dec.ctxIPC.ctx, dec.ipcURL) + } else { + i = ipc.NewClient(dec.ctxIPC.ctx, dec.ipcURL) + } gf := goffmpeg.New() if gf != nil { @@ -68,7 +73,7 @@ go dec.serializeImageInfo(imageChan, dataChan) gf.ActiveDecoder(func(d *[]byte, w, h *int) { - + fmt.Println("get a pic") if *w > 0 && *h > 0 { i := ImageInfo{*d, *w, *h, dec.cameraID} imageChan <- i @@ -78,7 +83,7 @@ }) gf.Run(dec.streamURL) - ipc.SendFromChannel(dataChan) + i.SendFromChannel(dataChan) } // Stop stop run decoder, must run in goroutine diff --git a/decoder/work/service/reciever.go b/decoder/work/service/reciever.go index 2ae6bfb..8c83c14 100644 --- a/decoder/work/service/reciever.go +++ b/decoder/work/service/reciever.go @@ -54,13 +54,18 @@ } // Run run task from ipc data -func (r *Reciever) Run() { - ipc := ipc.NewClient(r.ctxIPC.ctx, r.ipcURL) +func (r *Reciever) Run(asServer bool) { + var i *ipc.IPC + if asServer { + i = ipc.NewServer(r.ctxIPC.ctx, r.ipcURL) + } else { + i = ipc.NewClient(r.ctxIPC.ctx, r.ipcURL) + } dataChan := make(chan []byte) go unserilizeImageInfo(dataChan, r.chImage) - ipc.RecvToChannel(dataChan) + i.RecvToChannel(dataChan) } // Stop stop reciever, run in goroutine -- Gitblit v1.8.0