Update go2rtc candidates processing

cmd/webrtc/candidates.go

@@ -48,8 +48,8 @@ func GetCandidates() (candidates []string) {
 
 		candidates = append(
 			candidates,
-			webrtc.CandidateHostUDP(address.Host, address.Port),
+			webrtc.CandidateManualHostUDP(address.Host, address.Port),
-			webrtc.CandidateHostTCPPassive(address.Host, address.Port),
+			webrtc.CandidateManualHostTCPPassive(address.Host, address.Port),
 		)
 	}
 
@@ -90,19 +90,10 @@ func syncCanditates(answer string) (string, error) {
 
 	md := sd.MediaDescriptions[0]
 
-	_, end := md.Attribute("end-of-candidates")
-	if end {
-		md.Attributes = md.Attributes[:len(md.Attributes)-1]
-	}
-
 	for _, candidate := range GetCandidates() {
 		md.WithPropertyAttribute(candidate)
 	}
 
-	if end {
-		md.WithPropertyAttribute("end-of-candidates")
-	}
-
 	data, err := sd.Marshal()
 	if err != nil {
 		return "", err

cmd/webrtc/client.go

@@ -3,6 +3,7 @@ package webrtc
 import (
 	"errors"
 	"github.com/AlexxIT/go2rtc/cmd/api"
+	"github.com/AlexxIT/go2rtc/pkg/core"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"github.com/AlexxIT/go2rtc/pkg/webrtc"
 	"github.com/gorilla/websocket"
@@ -45,6 +46,8 @@ func asyncClient(url string) (streamer.Producer, error) {
 		return nil, err
 	}
 
+	var sendOffer core.Waiter
+
 	prod := webrtc.NewConn(pc)
 	prod.Listen(func(msg any) {
 		switch msg := msg.(type) {
@@ -52,12 +55,12 @@ func asyncClient(url string) (streamer.Producer, error) {
 			_ = ws.Close()
 
 		case *pion.ICECandidate:
-			if msg != nil {
+			sendOffer.Wait()
+
 			s := msg.ToJSON().Candidate
 			log.Trace().Str("candidate", s).Msg("[webrtc] local")
 			_ = ws.WriteJSON(&api.Message{Type: "webrtc/candidate", Value: s})
 		}
-		}
 	})
 
 	medias := []*streamer.Media{
@@ -77,6 +80,8 @@ func asyncClient(url string) (streamer.Producer, error) {
 		return nil, err
 	}
 
+	sendOffer.Done()
+
 	// 5. Get answer
 	if err = ws.ReadJSON(msg); err != nil {
 		return nil, err

cmd/webrtc/init.go

@@ -111,10 +111,6 @@ func asyncHandler(tr *api.Transport, msg *api.Message) error {
 			}
 
 		case *pion.ICECandidate:
-			if msg == nil {
-				return
-			}
-
 			sendAnswer.Wait()
 
 			s := msg.ToJSON().Candidate

pkg/core/helpers.go

@@ -51,3 +51,21 @@ func (w *Waiter) Done() {
 
 	w.mu.Unlock()
 }
+
+func (w *Waiter) WaitChan() <-chan struct{} {
+	var ch chan struct{}
+
+	w.mu.Lock()
+
+	if w.state >= 0 {
+		ch = make(chan struct{})
+		go func() {
+			w.Wait()
+			ch <- struct{}{}
+		}()
+	}
+
+	w.mu.Unlock()
+
+	return ch
+}

pkg/webrtc/conn.go

@@ -28,7 +28,10 @@ func NewConn(pc *webrtc.PeerConnection) *Conn {
 	c := &Conn{pc: pc}
 
 	pc.OnICECandidate(func(candidate *webrtc.ICECandidate) {
+		// last candidate will be empty
+		if candidate != nil {
 			c.Fire(candidate)
+		}
 	})
 
 	pc.OnDataChannel(func(channel *webrtc.DataChannel) {

pkg/webrtc/helpers.go

@@ -159,14 +159,22 @@ func MimeType(codec *streamer.Codec) string {
 	panic("not implemented")
 }
 
-const PriorityHost = (1 << 24) * uint32(126)
+// 4.1.2.2.  Guidelines for Choosing Type and Local Preferences
+// The RECOMMENDED values are 126 for host candidates, 100
+// for server reflexive candidates, 110 for peer reflexive candidates,
+// and 0 for relayed candidates.
+
+// We use new priority 120 for Manual Host. It is lower than real Host,
+// but more then any other candidates.
+
+const PriorityManualHost = (1 << 24) * uint32(120)
 const PriorityLocalUDP = (1 << 8) * uint32(65535)
 const PriorityLocalTCPPassive = (1 << 8) * uint32((1<<13)*4+8191)
 const PriorityComponentRTP = uint32(256 - ice.ComponentRTP)
 
-func CandidateHostUDP(host string, port int) string {
+func CandidateManualHostUDP(host string, port int) string {
 	foundation := crc32.ChecksumIEEE([]byte("host" + host + "udp4"))
-	priority := PriorityHost + PriorityLocalUDP + PriorityComponentRTP
+	priority := PriorityManualHost + PriorityLocalUDP + PriorityComponentRTP
 
 	// 1. Foundation
 	// 2. Component, always 1 because RTP
@@ -181,9 +189,9 @@ func CandidateHostUDP(host string, port int) string {
 	)
 }
 
-func CandidateHostTCPPassive(address string, port int) string {
+func CandidateManualHostTCPPassive(address string, port int) string {
 	foundation := crc32.ChecksumIEEE([]byte("host" + address + "tcp4"))
-	priority := PriorityHost + PriorityLocalTCPPassive + PriorityComponentRTP
+	priority := PriorityManualHost + PriorityLocalTCPPassive + PriorityComponentRTP
 
 	return fmt.Sprintf(
 		"candidate:%d 1 tcp %d %s %d typ host tcptype passive",

pkg/webrtc/webrtc_test.go

@@ -18,7 +18,7 @@ func TestCandidates(t *testing.T) {
 	}
 	cand, err := ice.NewCandidateHost(conf)
 	require.Nil(t, err)
-	assert.Equal(t, "candidate:"+cand.Marshal(), CandidateHostUDP(conf.Address, conf.Port))
+	assert.Equal(t, "candidate:"+cand.Marshal(), CandidateManualHostUDP(conf.Address, conf.Port))
 
 	conf = &ice.CandidateHostConfig{
 		Network:   "tcp",
@@ -29,7 +29,7 @@ func TestCandidates(t *testing.T) {
 	}
 	cand, err = ice.NewCandidateHost(conf)
 	require.Nil(t, err)
-	assert.Equal(t, "candidate:"+cand.Marshal(), CandidateHostTCPPassive(conf.Address, conf.Port))
+	assert.Equal(t, "candidate:"+cand.Marshal(), CandidateManualHostTCPPassive(conf.Address, conf.Port))
 }
 
 func TestPublicIP(t *testing.T) {