feat(sfu): Use two peer connections (#305)

* feat(sfu): Use two peer connections * feat(sfu): Fix dc api * feat(sfu): Move Peer to session * feat(sfu): Fix map extensions * fix proto * update trickles * Fix trickle * fix protos * grpc trickle target * set pion target to ext callback * fix subscribing issues * fix audio not setting twcc * Add callback to unmatched sdp * Increase debounce time * Update pion version * Fix tests * Fix race * simplify pending senders Co-authored-by: tarrencev <[email protected]>
ionorg · Nov 15, 2020 · 6fb9a74 · 6fb9a74
1 parent 3c81096
commit 6fb9a74
Show file tree

Hide file tree

Showing 20 changed files with 693 additions and 1,042 deletions.
diff --git a/Makefile b/Makefile
@@ -11,8 +11,8 @@ go_init:
 clean:
 	rm -rf bin
 
-build_proto:
-	docker run -v $(CURDIR):/workspace pionwebrtc/ion-sfu:latest-protoc protoc --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative cmd/signal/grpc/proto/sfu.proto
+protos:
+	docker build -t protoc-builder ./cmd/signal/grpc/proto && docker run -v $(CURDIR):/workspace protoc-builder protoc --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative cmd/signal/grpc/proto/sfu.proto
 
 
 build_grpc: go_init

diff --git a/cmd/signal/grpc/proto/Dockerfile b/cmd/signal/grpc/proto/Dockerfile
@@ -6,7 +6,7 @@ RUN go get google.golang.org/protobuf/cmd/protoc-gen-go
 
 RUN echo $GOPATH
 
-FROM alpine:3.12.0
+FROM alpine:3.12.1
 
 RUN apk --no-cache add protobuf
 

diff --git a/cmd/signal/grpc/proto/sfu.pb.go b/cmd/signal/grpc/proto/sfu.pb.go
diff --git a/cmd/signal/grpc/proto/sfu.proto b/cmd/signal/grpc/proto/sfu.proto
@@ -38,5 +38,10 @@ message JoinReply {
 }
 
 message Trickle {
-    string init = 1;
+    enum Target {
+        PUBLISHER = 0;
+        SUBSCRIBER = 1;
+    }
+    Target target = 1;
+    string init = 2;
 }
diff --git a/cmd/signal/grpc/server/server.go b/cmd/signal/grpc/server/server.go
@@ -104,15 +104,16 @@ func (s *SFUServer) Signal(stream pb.SFU_SignalServer) error {
 			}
 
 			// Notify user of new ice candidate
-			peer.OnIceCandidate = func(candidate *webrtc.ICECandidateInit) {
+			peer.OnIceCandidate = func(candidate *webrtc.ICECandidateInit, target int) {
 				bytes, err := json.Marshal(candidate)
 				if err != nil {
 					log.Errorf("OnIceCandidate error %s", err)
 				}
 				err = stream.Send(&pb.SignalReply{
 					Payload: &pb.SignalReply_Trickle{
 						Trickle: &pb.Trickle{
-							Init: string(bytes),
+							Init:   string(bytes),
+							Target: pb.Trickle_Target(target),
 						},
 					},
 				})
@@ -277,7 +278,7 @@ func (s *SFUServer) Signal(stream pb.SFU_SignalServer) error {
 				continue
 			}
 
-			err = peer.Trickle(candidate)
+			err = peer.Trickle(candidate, int(payload.Trickle.Target))
 			if err != nil {
 				switch err {
 				case sfu.ErrNoTransportEstablished:

diff --git a/cmd/signal/json-rpc/server/server.go b/cmd/signal/json-rpc/server/server.go
@@ -24,14 +24,15 @@ type Negotiation struct {
 
 // Trickle message sent when renegotiating the peer connection
 type Trickle struct {
+	Target    int                     `json:"target"`
 	Candidate webrtc.ICECandidateInit `json:"candidate"`
 }
 
 type JSONSignal struct {
-	sfu.Peer
+	*sfu.Peer
 }
 
-func NewJSONSignal(p sfu.Peer) *JSONSignal {
+func NewJSONSignal(p *sfu.Peer) *JSONSignal {
 	return &JSONSignal{p}
 }
 
@@ -66,8 +67,14 @@ func (p *JSONSignal) Handle(ctx context.Context, conn *jsonrpc2.Conn, req *jsonr
 			}
 
 		}
-		p.OnIceCandidate = func(candidate *webrtc.ICECandidateInit) {
-			if err := conn.Notify(ctx, "trickle", candidate); err != nil {
+		p.OnIceCandidate = func(candidate *webrtc.ICECandidateInit, target int) {
+			if err := conn.Notify(ctx, "trickle", struct {
+				Candidate *webrtc.ICECandidateInit `json:"candidate"`
+				Target    int                      `json:"target"`
+			}{
+				Candidate: candidate,
+				Target:    target,
+			}); err != nil {
 				log.Errorf("error sending ice candidate %s", err)
 			}
 		}
@@ -113,7 +120,7 @@ func (p *JSONSignal) Handle(ctx context.Context, conn *jsonrpc2.Conn, req *jsonr
 			break
 		}
 
-		err = p.Trickle(trickle.Candidate)
+		err = p.Trickle(trickle.Candidate, trickle.Target)
 		if err != nil {
 			replyError(err)
 		}

diff --git a/go.mod b/go.mod
@@ -5,23 +5,25 @@ go 1.13
 require (
 	github.com/bep/debounce v1.2.0
 	github.com/gammazero/deque v0.0.0-20201010052221-3932da5530cc
-	github.com/golang/protobuf v1.4.3 // indirect
+	github.com/golang/protobuf v1.4.3
 	github.com/gorilla/websocket v1.4.2
-	github.com/lucas-clemente/quic-go v0.18.1 // indirect
+	github.com/lucas-clemente/quic-go v0.19.1 // indirect
 	github.com/lucsky/cuid v1.0.2
 	github.com/marten-seemann/qtls-go1-15 v0.1.1 // indirect
+	github.com/matryer/moq v0.1.4 // indirect
+	github.com/pion/ice/v2 v2.0.10 // indirect
 	github.com/pion/ion-log v1.0.0
 	github.com/pion/rtcp v1.2.4
 	github.com/pion/rtp v1.6.1
 	github.com/pion/sdp/v3 v3.0.2
 	github.com/pion/turn/v2 v2.0.5 // indirect
-	github.com/pion/webrtc/v3 v3.0.0-beta.12.0.20201110054931-970a59f423f6
+	github.com/pion/webrtc/v3 v3.0.0-beta.12.0.20201115002753-64bbf7eea97d
 	github.com/sourcegraph/jsonrpc2 v0.0.0-20200429184054-15c2290dcb37
 	github.com/spf13/viper v1.7.1
 	github.com/stretchr/testify v1.6.1
-	golang.org/x/crypto v0.0.0-20201016220609-9e8e0b390897 // indirect
+	golang.org/x/crypto v0.0.0-20201112155050-0c6587e931a9 // indirect
 	golang.org/x/net v0.0.0-20201110031124-69a78807bb2b // indirect
-	golang.org/x/sys v0.0.0-20201109165425-215b40eba54c // indirect
+	golang.org/x/sys v0.0.0-20201113233024-12cec1faf1ba // indirect
 	google.golang.org/grpc v1.33.1
 	google.golang.org/protobuf v1.25.0
 	gopkg.in/ini.v1 v1.51.1 // indirect