Implement GetSelectedCandidatePair

Related to pion/webrtc#1713

agent.go

@@ -92,10 +92,10 @@ type Agent struct {
 	remotePwd        string
 	remoteCandidates map[NetworkType][]Candidate
 
-	checklist []*candidatePair
+	checklist []*CandidatePair
 	selector  pairCandidateSelector
 
-	selectedPair atomic.Value // *candidatePair
+	selectedPair atomic.Value // *CandidatePair
 
 	urls         []*URL
 	networkTypes []NetworkType
@@ -115,7 +115,7 @@ type Agent struct {
 	gatherCandidateCancel func()
 
 	chanCandidate     chan Candidate
-	chanCandidatePair chan *candidatePair
+	chanCandidatePair chan *CandidatePair
 	chanState         chan ConnectionState
 
 	loggerFactory logging.LoggerFactory
@@ -276,7 +276,7 @@ func NewAgent(config *AgentConfig) (*Agent, error) { //nolint:gocognit
 		chanTask:          make(chan task),
 		chanState:         make(chan ConnectionState),
 		chanCandidate:     make(chan Candidate),
-		chanCandidatePair: make(chan *candidatePair),
+		chanCandidatePair: make(chan *CandidatePair),
 		tieBreaker:        globalMathRandomGenerator.Uint64(),
 		lite:              config.Lite,
 		gatheringState:    GatheringStateNew,
@@ -379,9 +379,9 @@ func (a *Agent) OnCandidate(f func(Candidate)) error {
 	return nil
 }
 
-func (a *Agent) onSelectedCandidatePairChange(p *candidatePair) {
+func (a *Agent) onSelectedCandidatePairChange(p *CandidatePair) {
 	if h, ok := a.onSelectedCandidatePairChangeHdlr.Load().(func(Candidate, Candidate)); ok {
-		h(p.local, p.remote)
+		h(p.Local, p.Remote)
 	}
 }
 
@@ -559,11 +559,11 @@ func (a *Agent) updateConnectionState(newState ConnectionState) {
 	}
 }
 
-func (a *Agent) setSelectedPair(p *candidatePair) {
+func (a *Agent) setSelectedPair(p *CandidatePair) {
 	a.log.Tracef("Set selected candidate pair: %s", p)
 
 	if p == nil {
-		var nilPair *candidatePair
+		var nilPair *CandidatePair
 		a.selectedPair.Store(nilPair)
 		return
 	}
@@ -605,14 +605,14 @@ func (a *Agent) pingAllCandidates() {
 			a.log.Tracef("max requests reached for pair %s, marking it as failed\n", p)
 			p.state = CandidatePairStateFailed
 		} else {
-			a.selector.PingCandidate(p.local, p.remote)
+			a.selector.PingCandidate(p.Local, p.Remote)
 			p.bindingRequestCount++
 		}
 	}
 }
 
-func (a *Agent) getBestAvailableCandidatePair() *candidatePair {
-	var best *candidatePair
+func (a *Agent) getBestAvailableCandidatePair() *CandidatePair {
+	var best *CandidatePair
 	for _, p := range a.checklist {
 		if p.state == CandidatePairStateFailed {
 			continue
@@ -620,15 +620,15 @@ func (a *Agent) getBestAvailableCandidatePair() *candidatePair {
 
 		if best == nil {
 			best = p
-		} else if best.Priority() < p.Priority() {
+		} else if best.priority() < p.priority() {
 			best = p
 		}
 	}
 	return best
 }
 
-func (a *Agent) getBestValidCandidatePair() *candidatePair {
-	var best *candidatePair
+func (a *Agent) getBestValidCandidatePair() *CandidatePair {
+	var best *CandidatePair
 	for _, p := range a.checklist {
 		if p.state != CandidatePairStateSucceeded {
 			continue
@@ -636,22 +636,22 @@ func (a *Agent) getBestValidCandidatePair() *candidatePair {
 
 		if best == nil {
 			best = p
-		} else if best.Priority() < p.Priority() {
+		} else if best.priority() < p.priority() {
 			best = p
 		}
 	}
 	return best
 }
 
-func (a *Agent) addPair(local, remote Candidate) *candidatePair {
+func (a *Agent) addPair(local, remote Candidate) *CandidatePair {
 	p := newCandidatePair(local, remote, a.isControlling)
 	a.checklist = append(a.checklist, p)
 	return p
 }
 
-func (a *Agent) findPair(local, remote Candidate) *candidatePair {
+func (a *Agent) findPair(local, remote Candidate) *CandidatePair {
 	for _, p := range a.checklist {
-		if p.local.Equal(local) && p.remote.Equal(remote) {
+		if p.Local.Equal(local) && p.Remote.Equal(remote) {
 			return p
 		}
 	}
@@ -666,7 +666,7 @@ func (a *Agent) validateSelectedPair() bool {
 		return false
 	}
 
-	disconnectedTime := time.Since(selectedPair.remote.LastReceived())
+	disconnectedTime := time.Since(selectedPair.Remote.LastReceived())
 
 	// Only allow transitions to failed if a.failedTimeout is non-zero
 	totalTimeToFailure := a.failedTimeout
@@ -696,11 +696,11 @@ func (a *Agent) checkKeepalive() {
 	}
 
 	if (a.keepaliveInterval != 0) &&
-		((time.Since(selectedPair.local.LastSent()) > a.keepaliveInterval) ||
-			(time.Since(selectedPair.remote.LastReceived()) > a.keepaliveInterval)) {
+		((time.Since(selectedPair.Local.LastSent()) > a.keepaliveInterval) ||
+			(time.Since(selectedPair.Remote.LastReceived()) > a.keepaliveInterval)) {
 		// we use binding request instead of indication to support refresh consent schemas
 		// see https://tools.ietf.org/html/rfc7675
-		a.selector.PingCandidate(selectedPair.local, selectedPair.remote)
+		a.selector.PingCandidate(selectedPair.Local, selectedPair.Remote)
 	}
 }
 
@@ -1122,14 +1122,33 @@ func (a *Agent) validateNonSTUNTraffic(local Candidate, remote net.Addr) bool {
 	return atomic.LoadUint64(&isValidCandidate) == 1
 }
 
-func (a *Agent) getSelectedPair() *candidatePair {
-	selectedPair := a.selectedPair.Load()
+// GetSelectedCandidatePair returns the selected pair or nil if there is none
+func (a *Agent) GetSelectedCandidatePair() (*CandidatePair, error) {
+	selectedPair := a.getSelectedPair()
+	if selectedPair == nil {
+		return nil, nil
+	}
 
+	local, err := selectedPair.Local.copy()
+	if err != nil {
+		return nil, err
+	}
+
+	remote, err := selectedPair.Remote.copy()
+	if err != nil {
+		return nil, err
+	}
+
+	return &CandidatePair{Local: local, Remote: remote}, nil
+}
+
+func (a *Agent) getSelectedPair() *CandidatePair {
+	selectedPair := a.selectedPair.Load()
 	if selectedPair == nil {
 		return nil
 	}
 
-	return selectedPair.(*candidatePair)
+	return selectedPair.(*CandidatePair)
 }
 
 func (a *Agent) closeMulticastConn() {
@@ -1196,7 +1215,7 @@ func (a *Agent) Restart(ufrag, pwd string) error {
 		agent.remoteUfrag = ""
 		agent.remotePwd = ""
 		a.gatheringState = GatheringStateNew
-		a.checklist = make([]*candidatePair, 0)
+		a.checklist = make([]*CandidatePair, 0)
 		a.pendingBindingRequests = make([]bindingRequest, 0)
 		a.setSelectedPair(nil)
 		a.deleteAllCandidates()

agent_stats.go

@@ -13,8 +13,8 @@ func (a *Agent) GetCandidatePairsStats() []CandidatePairStats {
 		for _, cp := range agent.checklist {
 			stat := CandidatePairStats{
 				Timestamp:         time.Now(),
-				LocalCandidateID:  cp.local.ID(),
-				RemoteCandidateID: cp.remote.ID(),
+				LocalCandidateID:  cp.Local.ID(),
+				RemoteCandidateID: cp.Remote.ID(),
 				State:             cp.state,
 				Nominated:         cp.nominated,
 				// PacketsSent uint32

agent_test.go

@@ -139,7 +139,7 @@ func TestPairPriority(t *testing.T) {
 
 		p.state = CandidatePairStateSucceeded
 		bestPair := a.getBestValidCandidatePair()
-		if bestPair.String() != (&candidatePair{remote: remote, local: hostLocal}).String() {
+		if bestPair.String() != (&CandidatePair{Remote: remote, Local: hostLocal}).String() {
 			t.Fatalf("Unexpected bestPair %s (expected remote: %s)", bestPair, remote)
 		}
 	}
@@ -1702,3 +1702,60 @@ func TestNilCandidatePair(t *testing.T) {
 	a.setSelectedPair(nil)
 	assert.NoError(t, a.Close())
 }
+
+func TestGetSelectedCandidatePair(t *testing.T) {
+	report := test.CheckRoutines(t)
+	defer report()
+
+	lim := test.TimeOut(time.Second * 30)
+	defer lim.Stop()
+
+	wan, err := vnet.NewRouter(&vnet.RouterConfig{
+		CIDR:          "0.0.0.0/0",
+		LoggerFactory: logging.NewDefaultLoggerFactory(),
+	})
+	assert.NoError(t, err)
+
+	net := vnet.NewNet(&vnet.NetConfig{
+		StaticIPs: []string{"192.168.0.1"},
+	})
+	assert.NoError(t, wan.AddNet(net))
+
+	assert.NoError(t, wan.Start())
+
+	cfg := &AgentConfig{
+		NetworkTypes: supportedNetworkTypes(),
+		Net:          net,
+	}
+
+	aAgent, err := NewAgent(cfg)
+	assert.NoError(t, err)
+
+	bAgent, err := NewAgent(cfg)
+	assert.NoError(t, err)
+
+	aAgentPair, err := aAgent.GetSelectedCandidatePair()
+	assert.NoError(t, err)
+	assert.Nil(t, aAgentPair)
+
+	bAgentPair, err := bAgent.GetSelectedCandidatePair()
+	assert.NoError(t, err)
+	assert.Nil(t, bAgentPair)
+
+	connect(aAgent, bAgent)
+
+	aAgentPair, err = aAgent.GetSelectedCandidatePair()
+	assert.NoError(t, err)
+	assert.NotNil(t, aAgentPair)
+
+	bAgentPair, err = bAgent.GetSelectedCandidatePair()
+	assert.NoError(t, err)
+	assert.NotNil(t, bAgentPair)
+
+	assert.True(t, bAgentPair.Local.Equal(aAgentPair.Remote))
+	assert.True(t, bAgentPair.Remote.Equal(aAgentPair.Local))
+
+	assert.NoError(t, wan.Stop())
+	assert.NoError(t, aAgent.Close())
+	assert.NoError(t, bAgent.Close())
+}

candidate.go

@@ -62,6 +62,7 @@ type Candidate interface {
 	context() context.Context
 
 	close() error
+	copy() (Candidate, error)
 	seen(outbound bool)
 	start(a *Agent, conn net.PacketConn, initializedCh <-chan struct{})
 	writeTo(raw []byte, dst Candidate) (int, error)

candidate_base.go

@@ -387,6 +387,10 @@ func (c *candidateBase) context() context.Context {
 	return c
 }
 
+func (c *candidateBase) copy() (Candidate, error) {
+	return UnmarshalCandidate(c.Marshal())
+}
+
 // Marshal returns the string representation of the ICECandidate
 func (c *candidateBase) Marshal() string {
 	val := fmt.Sprintf("%s %d %s %d %s %d typ %s",
@@ -402,11 +406,11 @@ func (c *candidateBase) Marshal() string {
 		val += fmt.Sprintf(" tcptype %s", c.tcpType.String())
 	}
 
-	if c.RelatedAddress() != nil {
+	if r := c.RelatedAddress(); r != nil && r.Address != "" && r.Port != 0 {
 		val = fmt.Sprintf("%s raddr %s rport %d",
 			val,
-			c.RelatedAddress().Address,
-			c.RelatedAddress().Port)
+			r.Address,
+			r.Port)
 	}
 
 	return val

candidatepair.go

@@ -6,42 +6,43 @@ import (
 	"github.com/pion/stun"
 )
 
-func newCandidatePair(local, remote Candidate, controlling bool) *candidatePair {
-	return &candidatePair{
+func newCandidatePair(local, remote Candidate, controlling bool) *CandidatePair {
+	return &CandidatePair{
 		iceRoleControlling: controlling,
-		remote:             remote,
-		local:              local,
+		Remote:             remote,
+		Local:              local,
 		state:              CandidatePairStateWaiting,
 	}
 }
 
-// candidatePair represents a combination of a local and remote candidate
-type candidatePair struct {
+// CandidatePair is a combination of a
+// local and remote candidate
+type CandidatePair struct {
 	iceRoleControlling  bool
-	remote              Candidate
-	local               Candidate
+	Remote              Candidate
+	Local               Candidate
 	bindingRequestCount uint16
 	state               CandidatePairState
 	nominated           bool
 }
 
-func (p *candidatePair) String() string {
+func (p *CandidatePair) String() string {
 	if p == nil {
 		return ""
 	}
 
 	return fmt.Sprintf("prio %d (local, prio %d) %s <-> %s (remote, prio %d)",
-		p.Priority(), p.local.Priority(), p.local, p.remote, p.remote.Priority())
+		p.priority(), p.Local.Priority(), p.Local, p.Remote, p.Remote.Priority())
 }
 
-func (p *candidatePair) Equal(other *candidatePair) bool {
+func (p *CandidatePair) equal(other *CandidatePair) bool {
 	if p == nil && other == nil {
 		return true
 	}
 	if p == nil || other == nil {
 		return false
 	}
-	return p.local.Equal(other.local) && p.remote.Equal(other.remote)
+	return p.Local.Equal(other.Local) && p.Remote.Equal(other.Remote)
 }
 
 // RFC 5245 - 5.7.2.  Computing Pair Priority and Ordering Pairs
@@ -49,15 +50,14 @@ func (p *candidatePair) Equal(other *candidatePair) bool {
 // agent.  Let D be the priority for the candidate provided by the
 // controlled agent.
 // pair priority = 2^32*MIN(G,D) + 2*MAX(G,D) + (G>D?1:0)
-func (p *candidatePair) Priority() uint64 {
-	var g uint32
-	var d uint32
+func (p *CandidatePair) priority() uint64 {
+	var g, d uint32
 	if p.iceRoleControlling {
-		g = p.local.Priority()
-		d = p.remote.Priority()
+		g = p.Local.Priority()
+		d = p.Remote.Priority()
 	} else {
-		g = p.remote.Priority()
-		d = p.local.Priority()
+		g = p.Remote.Priority()
+		d = p.Local.Priority()
 	}
 
 	// Just implement these here rather
@@ -86,8 +86,8 @@ func (p *candidatePair) Priority() uint64 {
 	return (1<<32-1)*min(g, d) + 2*max(g, d) + cmp(g, d)
 }
 
-func (p *candidatePair) Write(b []byte) (int, error) {
-	return p.local.writeTo(b, p.remote)
+func (p *CandidatePair) Write(b []byte) (int, error) {
+	return p.Local.writeTo(b, p.Remote)
 }
 
 func (a *Agent) sendSTUN(msg *stun.Message, local, remote Candidate) {

candidatepair_test.go

@@ -44,7 +44,7 @@ func relayCandidate() *CandidateRelay {
 
 func TestCandidatePairPriority(t *testing.T) {
 	for _, test := range []struct {
-		Pair         *candidatePair
+		Pair         *CandidatePair
 		WantPriority uint64
 	}{
 		{
@@ -112,7 +112,7 @@ func TestCandidatePairPriority(t *testing.T) {
 			WantPriority: 72057593987596287,
 		},
 	} {
-		if got, want := test.Pair.Priority(), test.WantPriority; got != want {
+		if got, want := test.Pair.priority(), test.WantPriority; got != want {
 			t.Fatalf("CandidatePair(%v).Priority() = %d, want %d", test.Pair, got, want)
 		}
 	}
@@ -122,12 +122,12 @@ func TestCandidatePairEquality(t *testing.T) {
 	pairA := newCandidatePair(hostCandidate(), srflxCandidate(), true)
 	pairB := newCandidatePair(hostCandidate(), srflxCandidate(), false)
 
-	if !pairA.Equal(pairB) {
+	if !pairA.equal(pairB) {
 		t.Fatalf("Expected %v to equal %v", pairA, pairB)
 	}
 }
 
 func TestNilCandidatePairString(t *testing.T) {
-	var nilCandidatePair *candidatePair
+	var nilCandidatePair *CandidatePair
 	assert.Equal(t, nilCandidatePair.String(), "")
 }

selection.go

@@ -19,7 +19,7 @@ type pairCandidateSelector interface {
 type controllingSelector struct {
 	startTime     time.Time
 	agent         *Agent
-	nominatedPair *candidatePair
+	nominatedPair *CandidatePair
 	log           logging.LeveledLogger
 }
 
@@ -55,8 +55,8 @@ func (s *controllingSelector) ContactCandidates() {
 		s.nominatePair(s.nominatedPair)
 	default:
 		p := s.agent.getBestValidCandidatePair()
-		if p != nil && s.isNominatable(p.local) && s.isNominatable(p.remote) {
-			s.log.Tracef("Nominatable pair found, nominating (%s, %s)", p.local.String(), p.remote.String())
+		if p != nil && s.isNominatable(p.Local) && s.isNominatable(p.Remote) {
+			s.log.Tracef("Nominatable pair found, nominating (%s, %s)", p.Local.String(), p.Remote.String())
 			p.nominated = true
 			s.nominatedPair = p
 			s.nominatePair(p)
@@ -66,7 +66,7 @@ func (s *controllingSelector) ContactCandidates() {
 	}
 }
 
-func (s *controllingSelector) nominatePair(pair *candidatePair) {
+func (s *controllingSelector) nominatePair(pair *CandidatePair) {
 	// The controlling agent MUST include the USE-CANDIDATE attribute in
 	// order to nominate a candidate pair (Section 8.1.1).  The controlled
 	// agent MUST NOT include the USE-CANDIDATE attribute in a Binding
@@ -75,7 +75,7 @@ func (s *controllingSelector) nominatePair(pair *candidatePair) {
 		stun.NewUsername(s.agent.remoteUfrag+":"+s.agent.localUfrag),
 		UseCandidate(),
 		AttrControlling(s.agent.tieBreaker),
-		PriorityAttr(pair.local.Priority()),
+		PriorityAttr(pair.Local.Priority()),
 		stun.NewShortTermIntegrity(s.agent.remotePwd),
 		stun.Fingerprint,
 	)
@@ -84,8 +84,8 @@ func (s *controllingSelector) nominatePair(pair *candidatePair) {
 		return
 	}
 
-	s.log.Tracef("ping STUN (nominate candidate pair) from %s to %s\n", pair.local.String(), pair.remote.String())
-	s.agent.sendBindingRequest(msg, pair.local, pair.remote)
+	s.log.Tracef("ping STUN (nominate candidate pair) from %s to %s\n", pair.Local.String(), pair.Remote.String())
+	s.agent.sendBindingRequest(msg, pair.Local, pair.Remote)
 }
 
 func (s *controllingSelector) HandleBindingRequest(m *stun.Message, local, remote Candidate) {
@@ -102,9 +102,9 @@ func (s *controllingSelector) HandleBindingRequest(m *stun.Message, local, remot
 		bestPair := s.agent.getBestAvailableCandidatePair()
 		if bestPair == nil {
 			s.log.Tracef("No best pair available\n")
-		} else if bestPair.Equal(p) && s.isNominatable(p.local) && s.isNominatable(p.remote) {
+		} else if bestPair.equal(p) && s.isNominatable(p.Local) && s.isNominatable(p.Remote) {
 			s.log.Tracef("The candidate (%s, %s) is the best candidate available, marking it as nominated\n",
-				p.local.String(), p.remote.String())
+				p.Local.String(), p.Remote.String())
 			s.nominatedPair = p
 			s.nominatePair(p)
 		}

transport.go

@@ -116,7 +116,7 @@ func (c *Conn) LocalAddr() net.Addr {
 		return nil
 	}
 
-	return pair.local.addr()
+	return pair.Local.addr()
 }
 
 // RemoteAddr returns the remote address of the current selected pair or nil if there is none.
@@ -126,7 +126,7 @@ func (c *Conn) RemoteAddr() net.Addr {
 		return nil
 	}
 
-	return pair.remote.addr()
+	return pair.Remote.addr()
 }
 
 // SetDeadline is a stub

transport_test.go

@@ -185,13 +185,17 @@ func gatherAndExchangeCandidates(aAgent, bAgent *Agent) {
 	candidates, err := aAgent.GetLocalCandidates()
 	check(err)
 	for _, c := range candidates {
-		check(bAgent.AddRemoteCandidate(copyCandidate(c)))
+		candidateCopy, copyErr := c.copy()
+		check(copyErr)
+		check(bAgent.AddRemoteCandidate(candidateCopy))
 	}
 
 	candidates, err = bAgent.GetLocalCandidates()
 	check(err)
 	for _, c := range candidates {
-		check(aAgent.AddRemoteCandidate(copyCandidate(c)))
+		candidateCopy, copyErr := c.copy()
+		check(copyErr)
+		check(aAgent.AddRemoteCandidate(candidateCopy))
 	}
 }
 
@@ -283,14 +287,6 @@ func pipeWithTimeout(disconnectTimeout time.Duration, iceKeepalive time.Duration
 	return aConn, bConn
 }
 
-func copyCandidate(o Candidate) (c Candidate) {
-	c, err := UnmarshalCandidate(o.Marshal())
-	if err != nil {
-		panic(err)
-	}
-	return c
-}
-
 func onConnected() (func(ConnectionState), chan struct{}) {
 	done := make(chan struct{})
 	return func(state ConnectionState) {