diff --git a/agent.go b/agent.go index 9f1ac6a..958b6ba 100644 --- a/agent.go +++ b/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() diff --git a/agent_stats.go b/agent_stats.go index 18d9ed8..88c39f7 100644 --- a/agent_stats.go +++ b/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 diff --git a/agent_test.go b/agent_test.go index 7a6261a..240fbca 100644 --- a/agent_test.go +++ b/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()) +} diff --git a/candidate.go b/candidate.go index 701f8f8..c3049dd 100644 --- a/candidate.go +++ b/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) diff --git a/candidate_base.go b/candidate_base.go index 2f16a8a..c4f6f1b 100644 --- a/candidate_base.go +++ b/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 diff --git a/candidatepair.go b/candidatepair.go index 49dec37..79cab1a 100644 --- a/candidatepair.go +++ b/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) { diff --git a/candidatepair_test.go b/candidatepair_test.go index 1d817c1..119ab01 100644 --- a/candidatepair_test.go +++ b/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(), "") } diff --git a/selection.go b/selection.go index e0cfb10..7248414 100644 --- a/selection.go +++ b/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) } diff --git a/transport.go b/transport.go index d1c82ff..6d9aaff 100644 --- a/transport.go +++ b/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 diff --git a/transport_test.go b/transport_test.go index 97cbba0..a633e9f 100644 --- a/transport_test.go +++ b/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) {