pubsub/task.git

New file
			@@ -0,0 +1,3 @@
			[submodule "deliver"]
			path = deliver
			url = git://192.168.1.14/valib/deliver.git

New file
			@@ -1 +1 @@
			Subproject commit 0000000000000000000000000000000000000000
			Subproject commit 9a89af693b9336633bcac2a652c294f782e6b3b1

			@@ -6,6 +6,9 @@
			github.com/Microsoft/go-winio v0.4.12 // indirect
			github.com/golang/protobuf v1.3.1
			github.com/gorilla/websocket v1.4.0 // 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-20190418153312-f0ce4c0180be // indirect
			nanomsg.org/go-mangos v1.4.0
			)

			@@ -4,6 +4,12 @@
			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/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
			github.com/tmthrgd/go-sem v0.0.0-20160607101025-0214dbf53877 h1:n65+IT/xy5+trHm3Zpg9+j7IO4n8pBcPzvaKbMolW8U=
			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-20190418153312-f0ce4c0180be h1:mI+jhqkn68ybP0ORJqunXn+fq+Eeb4hHKqLQcFICjAc=
			golang.org/x/sys v0.0.0-20190418153312-f0ce4c0180be/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
			nanomsg.org/go-mangos v1.4.0 h1:pVRLnzXePdSbhWlWdSncYszTagERhMG5zK/vXYmbEdM=

			@@ -32,13 +32,21 @@

			for _, sdkid := range sdklist { // 创建sdk server
			url := fmt.Sprintf("%s%d", "tcp://192.168.1.124:", UrlPort)
			sdkid, socket, err := NewSdkSocketListen(deliver.PushPull, sdkid, url)
			sdkidser, socketser, err := NewSdkSocketListen(deliver.PushPull, sdkid, url)
			if err != nil {
			continue
			}
			go Send(sdkid, socket, SdkMap[sdkid])
			//break
			// Recv(socket)
			UrlPort++
			go Send(sdkidser, socketser, SdkMap[sdkid])

			url = fmt.Sprintf("%s%d", "tcp://192.168.1.124:", UrlPortR)
			_, socketdial, err := NewSdkSocketListen(deliver.PushPull, sdkid, url)
			if err != nil {
			continue
			}
			UrlPortR++
			go Recv(socketdial)

			}

			go es(SdkMap["es"])
			@@ -47,8 +55,8 @@

			//单独处理 es 主题的情况
			func es(sdkmsgchan chan *protomsg.SdkMessage) {
			for data := range sdkmsgchan {
			fmt.Println("this data is finish all sdk! ", data)
			for _ = range sdkmsgchan {
			fmt.Println("this data is finish all sdk! ")
			}

			}
			@@ -123,6 +131,7 @@
			}

			var UrlPort = 9000
			var UrlPortR = 9500

			// create server
			func NewSdkSocketListen(mode int, sdkid string, url string) (sid string, socket SocketContext, err error) {
			@@ -140,7 +149,24 @@
			}

			SocketManage[sdkid] = socket
			UrlPort++
			return sdkid, socket, nil
			}

			func NewSdkSocketDial(mode int, sdkid string, url string) (sid string, socket SocketContext, err error) {
			fmt.Println("url is: ", url)
			ctx, cancel := context.WithCancel(context.Background())

			socket.Context = ctx
			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")
			}

			SocketManage[sdkid] = socket
			return sdkid, socket, nil
			}

			@@ -159,6 +185,7 @@
			continue
			} else {
			err = proto.Unmarshal(msg, repsdkmsg)
			fmt.Println("receive len: ", len(msg))
			if err != nil {
			fmt.Println("unmarshal error: ", err)
			continue
			@@ -182,6 +209,7 @@
			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()
			@@ -189,12 +217,9 @@
			fmt.Println("send len of data: ", len(data))
			if err := socket.Sock.Send(data); err != nil {
			fmt.Println("failed send")
			continue
			}
			fmt.Printf("sdkid = %s ,send success:%d \n", sdkid, len(data))
			// go func(sock SocketContext) {
			// doOnce.Do(func() { Recv(sock) })

			// }(socket)
			}
			}
			}

Binary files differ

	.gitmodules	3 ●●●●● 补丁 \| 查看 \| 原始文档 \| blame \| 历史
	deliver @ 9a89af	2 ●●●●● 补丁 \| 查看 \| 原始文档 \| blame \| 历史
	go.mod	3 ●●●●● 补丁 \| 查看 \| 原始文档 \| blame \| 历史
	go.sum	6 ●●●●● 补丁 \| 查看 \| 原始文档 \| blame \| 历史
	sdk/sdk.go	47 ●●●●● 补丁 \| 查看 \| 原始文档 \| blame \| 历史
	test	补丁 \| 查看 \| 原始文档 \| blame \| 历史