Repository: pion/ice (go)

Issue:
Implement regular nomination
We need to stop using aggressive nomination. 

If we have multiple pairs this can cause us to have different choices on each side (since the last pair with `use-candidate` is chosen. 

Right now we just send every pair a 'USE-CANDIDATE' this means that each side may choose different candidates, all depending on when packets arrive.

Also then if you choose different candidates they will eventually die. ICE only does ping via SendIndications and we have no pongs. So each side is just sending pings and eventually each side dies since they haven't gotten any inbound traffic in 30 seconds.

Pull request:
Implement regular nomination
Fixes #28 

Patch:
diff --git a/agent.go b/agent.go
index 569c366..8b594e6 100644
--- a/agent.go
+++ b/agent.go
@@ -48,8 +48,9 @@ func (bp byPairPriority) Less(i, j int) bool {
 }
 
 type bindingRequest struct {
-	transactionID []byte
-	destination   net.Addr
+	transactionID  []byte
+	destination    net.Addr
+	isUseCandidate bool
 }
 
 // Agent represents the ICE agent
@@ -98,6 +99,7 @@ type Agent struct {
 	remotePwd        string
 	remoteCandidates map[NetworkType][]*Candidate
 
+	selector     pairCandidateSelector
 	selectedPair *candidatePair
 	validPairs   []*candidatePair
 
@@ -222,6 +224,7 @@ func NewAgent(config *AgentConfig) (*Agent, error) {
 	gatherCandidatesReflective(a, config.Urls, config.NetworkTypes)
 
 	go a.taskLoop()
+
 	return a, nil
 }
 
@@ -257,10 +260,18 @@ func (a *Agent) startConnectivityChecks(isControlling bool, remoteUfrag, remoteP
 	case remotePwd == "":
 		return ErrRemotePwdEmpty
 	}
+
 	a.haveStarted.Store(true)
 	a.log.Debugf("Started agent: isControlling? %t, remoteUfrag: %q, remotePwd: %q", isControlling, remoteUfrag, remotePwd)
 
 	return a.run(func(agent *Agent) {
+		if isControlling {
+			a.selector = &controllingSelector{agent: a, log: a.log}
+		} else {
+			a.selector = &controlledSelector{agent: a, log: a.log}
+		}
+		a.selector.Start()
+
 		agent.isControlling = isControlling
 		agent.remoteUfrag = remoteUfrag
 		agent.remotePwd = remotePwd
@@ -274,57 +285,6 @@ func (a *Agent) startConnectivityChecks(isControlling bool, remoteUfrag, remoteP
 	})
 }
 
-func (a *Agent) pingCandidate(local, remote *Candidate) {
-	var msg *stun.Message
-	var err error
-
-	// 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
-	// request.
-
-	transactionID := stun.GenerateTransactionID()
-	if a.isControlling {
-		msg, err = stun.Build(stun.ClassRequest, stun.MethodBinding, transactionID,
-			&stun.Username{Username: a.remoteUfrag + ":" + a.localUfrag},
-			&stun.UseCandidate{},
-			&stun.IceControlling{TieBreaker: a.tieBreaker},
-			&stun.Priority{Priority: local.Priority()},
-			&stun.MessageIntegrity{
-				Key: []byte(a.remotePwd),
-			},
-			&stun.Fingerprint{},
-		)
-	} else {
-		msg, err = stun.Build(stun.ClassRequest, stun.MethodBinding, transactionID,
-			&stun.Username{Username: a.remoteUfrag + ":" + a.localUfrag},
-			&stun.IceControlled{TieBreaker: a.tieBreaker},
-			&stun.Priority{Priority: local.Priority()},
-			&stun.MessageIntegrity{
-				Key: []byte(a.remotePwd),
-			},
-			&stun.Fingerprint{},
-		)
-	}
-
-	if err != nil {
-		a.log.Debug(err.Error())
-		return
-	}
-
-	a.log.Tracef("ping STUN from %s to %s\n", local.String(), remote.String())
-	if overflow := len(a.pendingBindingRequests) - maxPendingBindingRequests; overflow > 1 {
-		a.log.Debugf("Discarded %d pending binding requests, pendingBindingRequests is full", overflow)
-		a.pendingBindingRequests = a.pendingBindingRequests[overflow:]
-	}
-
-	a.pendingBindingRequests = append(a.pendingBindingRequests, bindingRequest{
-		transactionID: transactionID,
-		destination:   remote.addr(),
-	})
-	a.sendSTUN(msg, local, remote)
-}
-
 func (a *Agent) updateConnectionState(newState ConnectionState) {
 	if a.connectionState != newState {
 		a.log.Infof("Setting new connection state: %s", newState)
@@ -338,33 +298,52 @@ func (a *Agent) updateConnectionState(newState ConnectionState) {
 	}
 }
 
-func (a *Agent) setValidPair(local, remote *Candidate, selected, controlling bool) {
-	// TODO: avoid duplicates
-	p := newCandidatePair(local, remote, controlling)
-	a.log.Tracef("Found valid candidate pair: %s (selected? %t)", p, selected)
-
-	if selected {
-		// Notify when the selected pair changes
-		if !a.selectedPair.Equal(p) {
-			a.onSelectedCandidatePairChange(p)
+func (a *Agent) findValidPair(local, remote *Candidate) *candidatePair {
+	for _, p := range a.validPairs {
+		if p.local == local && p.remote == remote {
+			return p
 		}
-		a.selectedPair = p
-		a.validPairs = nil
-		// TODO: only set state to connected on selecting final pair?
-		a.updateConnectionState(ConnectionStateConnected)
-	} else {
-		// keep track of pairs with succesfull bindings since any of them
-		// can be used for communication until the final pair is selected:
-		// https://tools.ietf.org/html/draft-ietf-ice-rfc5245bis-20#section-12
-		a.validPairs = append(a.validPairs, p)
-		// Sort the candidate pairs by priority of the remotes
-		sort.Sort(byPairPriority{a.validPairs})
 	}
+	return nil
+}
+
+func (a *Agent) addValidPair(local, remote *Candidate) *candidatePair {
+	p := a.findValidPair(local, remote)
+	if p != nil {
+		a.log.Tracef("Candidate pair is already valid: %s", p)
+		return p
+	}
+
+	p = newCandidatePair(local, remote, a.isControlling)
+	a.log.Tracef("Found valid candidate pair: %s", p)
+
+	// keep track of pairs with succesfull bindings since any of them
+	// can be used for communication until the final pair is selected:
+	// https://tools.ietf.org/html/draft-ietf-ice-rfc5245bis-20#section-12
+	a.validPairs = append(a.validPairs, p)
+	return p
+}
+
+func (a *Agent) setSelectedPair(p *candidatePair) {
+	a.log.Tracef("Set selected candidate pair: %s", p)
+	// Notify when the selected pair changes
+	a.onSelectedCandidatePairChange(p)
+
+	a.selectedPair = p
+	a.updateConnectionState(ConnectionStateConnected)
 
 	// Signal connected
 	a.onConnectedOnce.Do(func() { close(a.onConnected) })
 }
 
+func (a *Agent) getBestValidPair() *candidatePair {
+	if len(a.validPairs) == 0 {
+		return nil
+	}
+	sort.Sort(byPairPriority{a.validPairs})
+	return a.validPairs[0]
+}
+
 // A task is a
 type task func(*Agent)
 
@@ -383,28 +362,29 @@ func (a *Agent) run(t task) error {
 }
 
 func (a *Agent) taskLoop() {
-	contactCandidates := func() {
-		if a.validateSelectedPair() {
-			a.log.Trace("checking keepalive")
-			a.checkKeepalive()
+	for {
+		if a.selector != nil {
+			select {
+			case <-a.forceCandidateContact:
+				a.selector.ContactCandidates()
+			case <-a.connectivityChan:
+				a.selector.ContactCandidates()
+			case t := <-a.taskChan:
+				// Run the task
+				t(a)
+
+			case <-a.done:
+				return
+			}
 		} else {
-			a.log.Trace("pinging all candidates")
-			a.pingAllCandidates()
-		}
-	}
+			select {
+			case t := <-a.taskChan:
+				// Run the task
+				t(a)
 
-	for {
-		select {
-		case <-a.forceCandidateContact:
-			contactCandidates()
-		case <-a.connectivityChan:
-			contactCandidates()
-		case t := <-a.taskChan:
-			// Run the task
-			t(a)
-
-		case <-a.done:
-			return
+			case <-a.done:
+				return
+			}
 		}
 	}
 }
@@ -449,7 +429,7 @@ func (a *Agent) pingAllCandidates() {
 
 			for _, localCandidate := range localCandidates {
 				for _, remoteCandidate := range remoteCandidates {
-					a.pingCandidate(localCandidate, remoteCandidate)
+					a.selector.PingCandidate(localCandidate, remoteCandidate)
 				}
 			}
 
@@ -573,6 +553,25 @@ func (a *Agent) findRemoteCandidate(networkType NetworkType, addr net.Addr) *Can
 	return nil
 }
 
+func (a *Agent) sendBindingRequest(m *stun.Message, local, remote *Candidate) {
+	a.log.Tracef("ping STUN from %s to %s\n", local.String(), remote.String())
+
+	if overflow := len(a.pendingBindingRequests) - (maxPendingBindingRequests - 1); overflow > 0 {
+		a.log.Debugf("Discarded %d pending binding requests, pendingBindingRequests is full", overflow)
+		a.pendingBindingRequests = a.pendingBindingRequests[overflow:]
+	}
+
+	_, useCandidate := m.GetOneAttribute(stun.AttrUseCandidate)
+
+	a.pendingBindingRequests = append(a.pendingBindingRequests, bindingRequest{
+		transactionID:  m.TransactionID,
+		destination:    remote.addr(),
+		isUseCandidate: useCandidate,
+	})
+
+	a.sendSTUN(m, local, remote)
+}
+
 func (a *Agent) sendBindingSuccess(m *stun.Message, local, remote *Candidate) {
 	base := remote
 	if out, err := stun.Build(stun.ClassSuccessResponse, stun.MethodBinding, m.TransactionID,
@@ -595,13 +594,12 @@ func (a *Agent) sendBindingSuccess(m *stun.Message, local, remote *Candidate) {
 
 // Assert that the passed TransactionID is in our pendingBindingRequests and returns the destination
 // If the bindingRequest was valid remove it from our pending cache
-func (a *Agent) handleInboundBindingSuccess(id []byte) (bool, net.Addr) {
+func (a *Agent) handleInboundBindingSuccess(id []byte) (bool, *bindingRequest) {
 	for i := range a.pendingBindingRequests {
 		if bytes.Equal(a.pendingBindingRequests[i].transactionID, id) {
-			defer func() {
-				a.pendingBindingRequests = append(a.pendingBindingRequests[:i], a.pendingBindingRequests[i+1:]...)
-			}()
-			return true, a.pendingBindingRequests[i].destination
+			validBindingRequest := a.pendingBindingRequests[i]
+			a.pendingBindingRequests = append(a.pendingBindingRequests[:i], a.pendingBindingRequests[i+1:]...)
+			return true, &validBindingRequest
 		}
 	}
 	return false, nil
@@ -612,7 +610,9 @@ func (a *Agent) handleInbound(m *stun.Message, local *Candidate, remote net.Addr
 	var err error
 	if m == nil || local == nil {
 		return
-	} else if m.Method != stun.MethodBinding || !(m.Class == stun.ClassSuccessResponse || m.Class == stun.ClassRequest) {
+	}
+
+	if m.Method != stun.MethodBinding || !(m.Class == stun.ClassSuccessResponse || m.Class == stun.ClassRequest) {
 		a.log.Tracef("unhandled STUN from %s to %s class(%s) method(%s)", remote.String(), local.String(), m.Method.String(), m.Class.String())
 		return
 	}
@@ -639,26 +639,12 @@ func (a *Agent) handleInbound(m *stun.Message, local *Candidate, remote net.Addr
 			return
 		}
 
-		ok, transactionAddr := a.handleInboundBindingSuccess(m.TransactionID)
-		if !ok {
-			a.log.Errorf("discard message from (%s), invalid TransactionID %s", remote, m.TransactionID)
-			return
-		}
-
-		// Assert that NAT is not symmetric
-		// https://tools.ietf.org/html/rfc8445#section-7.2.5.2.1
-		if !addrEqual(transactionAddr, remote) {
-			a.log.Debugf("discard message: transaction source and destination does not match expected(%s), actual(%s)", transactionAddr, remote)
-			return
-		} else if remoteCandidate == nil { // Should fail previous check, better to be safe though
+		if remoteCandidate == nil {
 			a.log.Warnf("discard success message from (%s), no such remote", remote)
 			return
 		}
 
-		a.log.Tracef("inbound STUN (SuccessResponse) from %s to %s", remote.String(), local.String())
-
-		// Remember the working pair and select it when receiving a success response
-		a.setValidPair(local, remoteCandidate, true, true)
+		a.selector.HandleSucessResponse(m, local, remoteCandidate, remote)
 	} else {
 		if err = assertInboundUsername(m, a.localUfrag+":"+a.remoteUfrag); err != nil {
 			a.log.Warnf("discard message from (%s), %v", remote, err)
@@ -688,8 +674,7 @@ func (a *Agent) handleInbound(m *stun.Message, local *Candidate, remote net.Addr
 
 		a.log.Tracef("inbound STUN (Request) from %s to %s", remote.String(), local.String())
 
-		// Send success response
-		a.sendBindingSuccess(m, local, remoteCandidate)
+		a.selector.HandleBindingRequest(m, local, remoteCandidate)
 	}
 
 	remoteCandidate.seen(false)
@@ -707,7 +692,7 @@ func (a *Agent) noSTUNSeen(local *Candidate, remote net.Addr) bool {
 	return true
 }
 
-func (a *Agent) getBestPair() (*candidatePair, error) {
+func (a *Agent) getSelectedPair() (*candidatePair, error) {
 	res := make(chan *candidatePair)
 
 	err := a.run(func(agent *Agent) {
@@ -715,10 +700,6 @@ func (a *Agent) getBestPair() (*candidatePair, error) {
 			res <- agent.selectedPair
 			return
 		}
-		for _, p := range agent.validPairs {
-			res <- p
-			return
-		}
 		res <- nil
 	})
 
diff --git a/selection.go b/selection.go
new file mode 100644
index 0000000..71fd8ba
--- /dev/null
+++ b/selection.go
@@ -0,0 +1,220 @@
+package ice
+
+import (
+	"net"
+
+	"github.com/pion/logging"
+	"github.com/pion/stun"
+)
+
+type pairCandidateSelector interface {
+	Start()
+	ContactCandidates()
+	PingCandidate(local, remote *Candidate)
+	HandleSucessResponse(m *stun.Message, local, remote *Candidate, remoteAddr net.Addr)
+	HandleBindingRequest(m *stun.Message, local, remote *Candidate)
+}
+
+type controllingSelector struct {
+	agent         *Agent
+	nominatedPair *candidatePair
+	log           logging.LeveledLogger
+}
+
+func (s *controllingSelector) Start() {
+}
+
+func (s *controllingSelector) ContactCandidates() {
+	switch {
+	case s.agent.selectedPair != nil:
+		if s.agent.validateSelectedPair() {
+			s.log.Trace("checking keepalive")
+			s.agent.checkKeepalive()
+		}
+	case s.nominatedPair != nil:
+		s.nominatePair(s.nominatedPair)
+	default:
+		s.log.Trace("pinging all candidates")
+		s.agent.pingAllCandidates()
+	}
+}
+
+func (s *controllingSelector) nominatePair(pair *candidatePair) {
+	transactionID := stun.GenerateTransactionID()
+
+	// 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
+	// request.
+	msg, err := stun.Build(stun.ClassRequest, stun.MethodBinding, transactionID,
+		&stun.Username{Username: s.agent.remoteUfrag + ":" + s.agent.localUfrag},
+		&stun.UseCandidate{},
+		&stun.IceControlling{TieBreaker: s.agent.tieBreaker},
+		&stun.Priority{Priority: pair.local.Priority()},
+		&stun.MessageIntegrity{
+			Key: []byte(s.agent.remotePwd),
+		},
+		&stun.Fingerprint{},
+	)
+
+	if err != nil {
+		s.log.Error(err.Error())
+		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)
+}
+
+func (s *controllingSelector) HandleBindingRequest(m *stun.Message, local, remote *Candidate) {
+	s.agent.sendBindingSuccess(m, local, remote)
+
+	p := s.agent.findValidPair(local, remote)
+	if p != nil && s.nominatedPair == nil && s.agent.selectedPair == nil {
+		s.nominatedPair = p
+		s.nominatePair(p)
+	}
+}
+
+func (s *controllingSelector) HandleSucessResponse(m *stun.Message, local, remote *Candidate, remoteAddr net.Addr) {
+	ok, pendingRequest := s.agent.handleInboundBindingSuccess(m.TransactionID)
+	if !ok {
+		s.log.Errorf("discard message from (%s), invalid TransactionID %s", remote, m.TransactionID)
+		return
+	}
+
+	transactionAddr := pendingRequest.destination
+
+	// Assert that NAT is not symmetric
+	// https://tools.ietf.org/html/rfc8445#section-7.2.5.2.1
+	if !addrEqual(transactionAddr, remoteAddr) {
+		s.log.Debugf("discard message: transaction source and destination does not match expected(%s), actual(%s)", transactionAddr, remote)
+		return
+	}
+
+	s.log.Tracef("inbound STUN (SuccessResponse) from %s to %s", remote.String(), local.String())
+	p := s.agent.addValidPair(local, remote)
+
+	if pendingRequest.isUseCandidate {
+		s.agent.setSelectedPair(p)
+	}
+}
+
+func (s *controllingSelector) PingCandidate(local, remote *Candidate) {
+	transactionID := stun.GenerateTransactionID()
+
+	msg, err := stun.Build(stun.ClassRequest, stun.MethodBinding, transactionID,
+		&stun.Username{Username: s.agent.remoteUfrag + ":" + s.agent.localUfrag},
+		&stun.IceControlling{TieBreaker: s.agent.tieBreaker},
+		&stun.Priority{Priority: local.Priority()},
+		&stun.MessageIntegrity{
+			Key: []byte(s.agent.remotePwd),
+		},
+		&stun.Fingerprint{},
+	)
+
+	if err != nil {
+		s.log.Error(err.Error())
+		return
+	}
+
+	s.agent.sendBindingRequest(msg, local, remote)
+}
+
+type controlledSelector struct {
+	agent *Agent
+	log   logging.LeveledLogger
+}
+
+func (s *controlledSelector) Start() {}
+
+func (s *controlledSelector) ContactCandidates() {
+	if s.agent.selectedPair != nil {
+		if s.agent.validateSelectedPair() {
+			s.log.Trace("checking keepalive")
+			s.agent.checkKeepalive()
+		}
+	} else {
+		s.log.Trace("pinging all candidates")
+		s.agent.pingAllCandidates()
+	}
+}
+
+func (s *controlledSelector) PingCandidate(local, remote *Candidate) {
+	transactionID := stun.GenerateTransactionID()
+
+	msg, err := stun.Build(stun.ClassRequest, stun.MethodBinding, transactionID,
+		&stun.Username{Username: s.agent.remoteUfrag + ":" + s.agent.localUfrag},
+		&stun.IceControlled{TieBreaker: s.agent.tieBreaker},
+		&stun.Priority{Priority: local.Priority()},
+		&stun.MessageIntegrity{
+			Key: []byte(s.agent.remotePwd),
+		},
+		&stun.Fingerprint{},
+	)
+
+	if err != nil {
+		s.log.Error(err.Error())
+		return
+	}
+
+	s.agent.sendBindingRequest(msg, local, remote)
+}
+
+func (s *controlledSelector) HandleSucessResponse(m *stun.Message, local, remote *Candidate, remoteAddr net.Addr) {
+	// TODO according to the standard we should specifically answer a failed nomination:
+	// https://tools.ietf.org/html/rfc8445#section-7.3.1.5
+	// If the controlled agent does not accept the request from the
+	// controlling agent, the controlled agent MUST reject the nomination
+	// request with an appropriate error code response (e.g., 400)
+	// [RFC5389].
+
+	ok, pendingRequest := s.agent.handleInboundBindingSuccess(m.TransactionID)
+	if !ok {
+		s.log.Errorf("discard message from (%s), invalid TransactionID %s", remote, m.TransactionID)
+		return
+	}
+
+	transactionAddr := pendingRequest.destination
+
+	// Assert that NAT is not symmetric
+	// https://tools.ietf.org/html/rfc8445#section-7.2.5.2.1
+	if !addrEqual(transactionAddr, remoteAddr) {
+		s.log.Debugf("discard message: transaction source and destination does not match expected(%s), actual(%s)", transactionAddr, remote)
+		return
+	}
+
+	s.log.Tracef("inbound STUN (SuccessResponse) from %s to %s", remote.String(), local.String())
+	s.agent.addValidPair(local, remote)
+}
+
+func (s *controlledSelector) HandleBindingRequest(m *stun.Message, local, remote *Candidate) {
+	_, useCandidate := m.GetOneAttribute(stun.AttrUseCandidate)
+
+	if useCandidate {
+		// https://tools.ietf.org/html/rfc8445#section-7.3.1.5
+		p := s.agent.findValidPair(local, remote)
+
+		if p != nil {
+			// If the state of this pair is Succeeded, it means that the check
+			// previously sent by this pair produced a successful response and
+			// generated a valid pair (Section 7.2.5.3.2).  The agent sets the
+			// nominated flag value of the valid pair to true.
+			s.agent.setSelectedPair(p)
+			s.agent.sendBindingSuccess(m, local, remote)
+		} else {
+			// If the received Binding request triggered a new check to be
+			// enqueued in the triggered-check queue (Section 7.3.1.4), once the
+			// check is sent and if it generates a successful response, and
+			// generates a valid pair, the agent sets the nominated flag of the
+			// pair to true.  If the request fails (Section 7.2.5.2), the agent
+			// MUST remove the candidate pair from the valid list, set the
+			// candidate pair state to Failed, and set the checklist state to
+			// Failed.
+			s.PingCandidate(local, remote)
+		}
+	} else {
+		s.agent.sendBindingSuccess(m, local, remote)
+		s.PingCandidate(local, remote)
+	}
+}
diff --git a/transport.go b/transport.go
index 4253a43..a74cefb 100644
--- a/transport.go
+++ b/transport.go
@@ -75,7 +75,7 @@ func (c *Conn) Write(p []byte) (int, error) {
 		return 0, errors.New("the ICE conn can't write STUN messages")
 	}
 
-	pair, err := c.agent.getBestPair()
+	pair, err := c.agent.getSelectedPair()
 	if err != nil {
 		return 0, err
 	}


Test patch:
diff --git a/agent_test.go b/agent_test.go
index 0dd945c..654a353 100644
--- a/agent_test.go
+++ b/agent_test.go
@@ -37,16 +37,12 @@ func TestPairSearch(t *testing.T) {
 		t.Fatalf("TestPairSearch is only a valid test if a.validPairs is empty on construction")
 	}
 
-	cp, err := a.getBestPair()
+	cp := a.getBestValidPair()
 
 	if cp != nil {
 		t.Fatalf("No Candidate pairs should exist")
 	}
 
-	if err == nil {
-		t.Fatalf("An error should have been reported (with no available candidate pairs)")
-	}
-
 	err = a.Close()
 
 	if err != nil {
@@ -112,11 +108,8 @@ func TestPairPriority(t *testing.T) {
 	}
 
 	for _, remote := range []*Candidate{relayRemote, srflxRemote, prflxRemote, hostRemote} {
-		a.setValidPair(hostLocal, remote, false, false)
-		bestPair, err := a.getBestPair()
-		if err != nil {
-			t.Fatalf("Failed to get best candidate pair: %s", err)
-		}
+		a.addValidPair(hostLocal, remote)
+		bestPair := a.getBestValidPair()
 		if bestPair.String() != (&candidatePair{remote: remote, local: hostLocal}).String() {
 			t.Fatalf("Unexpected bestPair %s (expected remote: %s)", bestPair, remote)
 		}
@@ -163,23 +156,14 @@ func TestOnSelectedCandidatePairChange(t *testing.T) {
 
 	// select the pair
 	if err = a.run(func(agent *Agent) {
-		agent.setValidPair(hostLocal, relayRemote, true, false)
+		p := newCandidatePair(hostLocal, relayRemote, false)
+		agent.setSelectedPair(p)
 	}); err != nil {
 		t.Fatalf("Failed to setValidPair(): %s", err)
 	}
+
 	// ensure that the callback fired on setting the pair
 	<-callbackCalled
-	// set the same pair; this should not invoke the callback
-	// if the callback is invoked now it will panic due
-	// to second close of the channel
-	if err = a.run(func(agent *Agent) {
-		agent.setValidPair(hostLocal, relayRemote, true, false)
-	}); err != nil {
-		t.Fatalf("Failed to setValidPair(): %s", err)
-	}
-	if err := a.Close(); err != nil {
-		t.Fatalf("Error on agent.Close(): %s", err)
-	}
 }
 
 type BadAddr struct{}
@@ -191,6 +175,18 @@ func (ba *BadAddr) String() string {
 	return "yyy"
 }
 
+func runAgentTest(t *testing.T, config *AgentConfig, task func(a *Agent)) {
+	a, err := NewAgent(config)
+
+	if err != nil {
+		t.Fatalf("Error constructing ice.Agent")
+	}
+
+	if err := a.run(task); err != nil {
+		t.Fatalf("Agent run failure: %v", err)
+	}
+}
+
 func TestHandlePeerReflexive(t *testing.T) {
 	// Limit runtime in case of deadlocks
 	lim := test.TimeOut(time.Second * 2)
@@ -198,127 +194,120 @@ func TestHandlePeerReflexive(t *testing.T) {
 
 	t.Run("UDP pflx candidate from handleInbound()", func(t *testing.T) {
 		var config AgentConfig
-		a, err := NewAgent(&config)
-
-		if err != nil {
-			t.Fatalf("Error constructing ice.Agent")
-		}
-
-		ip := net.ParseIP("192.168.0.2")
-		local, err := NewCandidateHost("udp", ip, 777, 1)
-		local.conn = &mockPacketConn{}
-		if err != nil {
-			t.Fatalf("failed to create a new candidate: %v", err)
-		}
-
-		remote := &net.UDPAddr{IP: net.ParseIP("172.17.0.3"), Port: 999}
-
-		msg, err := stun.Build(stun.ClassRequest, stun.MethodBinding, stun.GenerateTransactionID(),
-			&stun.Username{Username: a.localUfrag + ":" + a.remoteUfrag},
-			&stun.UseCandidate{},
-			&stun.IceControlling{TieBreaker: a.tieBreaker},
-			&stun.Priority{Priority: local.Priority()},
-			&stun.MessageIntegrity{
-				Key: []byte(a.localPwd),
-			},
-			&stun.Fingerprint{},
-		)
-		if err != nil {
-			t.Fatal(err)
-		}
-
-		a.handleInbound(msg, local, remote)
-
-		// length of remote candidate list must be one now
-		if len(a.remoteCandidates) != 1 {
-			t.Fatal("failed to add a network type to the remote candidate list")
-		}
-
-		// length of remote candidate list for a network type must be 1
-		set := a.remoteCandidates[local.NetworkType]
-		if len(set) != 1 {
-			t.Fatal("failed to add prflx candidate to remote candidate list")
-		}
-
-		c := set[0]
-
-		if c.Type != CandidateTypePeerReflexive {
-			t.Fatal("candidate type must be prflx")
-		}
-
-		if !c.IP.Equal(net.ParseIP("172.17.0.3")) {
-			t.Fatal("IP address mismatch")
-		}
-
-		if c.Port != 999 {
-			t.Fatal("Port number mismatch")
-		}
-
-		err = a.Close()
-		if err != nil {
-			t.Fatalf("Close agent emits error %v", err)
-		}
+		runAgentTest(t, &config, func(a *Agent) {
+			a.selector = &controllingSelector{agent: a, log: a.log}
+			ip := net.ParseIP("192.168.0.2")
+			local, err := NewCandidateHost("udp", ip, 777, 1)
+			local.conn = &mockPacketConn{}
+			if err != nil {
+				t.Fatalf("failed to create a new candidate: %v", err)
+			}
+
+			remote := &net.UDPAddr{IP: net.ParseIP("172.17.0.3"), Port: 999}
+
+			msg, err := stun.Build(stun.ClassRequest, stun.MethodBinding, stun.GenerateTransactionID(),
+				&stun.Username{Username: a.localUfrag + ":" + a.remoteUfrag},
+				&stun.UseCandidate{},
+				&stun.IceControlling{TieBreaker: a.tieBreaker},
+				&stun.Priority{Priority: local.Priority()},
+				&stun.MessageIntegrity{
+					Key: []byte(a.localPwd),
+				},
+				&stun.Fingerprint{},
+			)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			a.handleInbound(msg, local, remote)
+
+			// length of remote candidate list must be one now
+			if len(a.remoteCandidates) != 1 {
+				t.Fatal("failed to add a network type to the remote candidate list")
+			}
+
+			// length of remote candidate list for a network type must be 1
+			set := a.remoteCandidates[local.NetworkType]
+			if len(set) != 1 {
+				t.Fatal("failed to add prflx candidate to remote candidate list")
+			}
+
+			c := set[0]
+
+			if c.Type != CandidateTypePeerReflexive {
+				t.Fatal("candidate type must be prflx")
+			}
+
+			if !c.IP.Equal(net.ParseIP("172.17.0.3")) {
+				t.Fatal("IP address mismatch")
+			}
+
+			if c.Port != 999 {
+				t.Fatal("Port number mismatch")
+			}
+
+			err = a.Close()
+			if err != nil {
+				t.Fatalf("Close agent emits error %v", err)
+			}
+		})
 	})
 
 	t.Run("Bad network type with handleInbound()", func(t *testing.T) {
 		var config AgentConfig
-		a, err := NewAgent(&config)
-
-		if err != nil {
-			t.Fatal("Error constructing ice.Agent")
-		}
-
-		ip := net.ParseIP("192.168.0.2")
-		local, err := NewCandidateHost("tcp", ip, 777, 1)
-		if err != nil {
-			t.Fatalf("failed to create a new candidate: %v", err)
-		}
-
-		remote := &BadAddr{}
-
-		a.handleInbound(nil, local, remote)
-
-		if len(a.remoteCandidates) != 0 {
-			t.Fatal("bad address should not be added to the remote candidate list")
-		}
-
-		err = a.Close()
-		if err != nil {
-			t.Fatalf("Close agent emits error %v", err)
-		}
+		runAgentTest(t, &config, func(a *Agent) {
+			a.selector = &controllingSelector{agent: a, log: a.log}
+			ip := net.ParseIP("192.168.0.2")
+			local, err := NewCandidateHost("tcp", ip, 777, 1)
+			if err != nil {
+				t.Fatalf("failed to create a new candidate: %v", err)
+			}
+
+			remote := &BadAddr{}
+
+			a.handleInbound(nil, local, remote)
+
+			if len(a.remoteCandidates) != 0 {
+				t.Fatal("bad address should not be added to the remote candidate list")
+			}
+
+			err = a.Close()
+			if err != nil {
+				t.Fatalf("Close agent emits error %v", err)
+			}
+		})
 	})
 
 	t.Run("Success from unknown remote, prflx candidate MUST only be created via Binding Request", func(t *testing.T) {
-		a, err := NewAgent(&AgentConfig{})
-		if err != nil {
-			t.Fatalf("Error constructing ice.Agent")
-		}
-
-		a.pendingBindingRequests = []bindingRequest{
-			{[]byte("ABC"), &net.UDPAddr{}},
-		}
-
-		local, err := NewCandidateHost("udp", net.ParseIP("192.168.0.2"), 777, 1)
-		local.conn = &mockPacketConn{}
-		if err != nil {
-			t.Fatalf("failed to create a new candidate: %v", err)
-		}
-
-		remote := &net.UDPAddr{IP: net.ParseIP("172.17.0.3"), Port: 999}
-		msg, err := stun.Build(stun.ClassSuccessResponse, stun.MethodBinding, []byte("ABC"),
-			&stun.MessageIntegrity{
-				Key: []byte(a.remotePwd),
-			},
-			&stun.Fingerprint{},
-		)
-		if err != nil {
-			t.Fatal(err)
-		}
-
-		a.handleInbound(msg, local, remote)
-		if len(a.remoteCandidates) != 0 {
-			t.Fatal("unknown remote was able to create a candidate")
-		}
+		var config AgentConfig
+		runAgentTest(t, &config, func(a *Agent) {
+			a.selector = &controllingSelector{agent: a, log: a.log}
+			a.pendingBindingRequests = []bindingRequest{
+				{[]byte("ABC"), &net.UDPAddr{}, false},
+			}
+
+			local, err := NewCandidateHost("udp", net.ParseIP("192.168.0.2"), 777, 1)
+			local.conn = &mockPacketConn{}
+			if err != nil {
+				t.Fatalf("failed to create a new candidate: %v", err)
+			}
+
+			remote := &net.UDPAddr{IP: net.ParseIP("172.17.0.3"), Port: 999}
+			msg, err := stun.Build(stun.ClassSuccessResponse, stun.MethodBinding, []byte("ABC"),
+				&stun.MessageIntegrity{
+					Key: []byte(a.remotePwd),
+				},
+				&stun.Fingerprint{},
+			)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			a.handleInbound(msg, local, remote)
+			if len(a.remoteCandidates) != 0 {
+				t.Fatal("unknown remote was able to create a candidate")
+			}
+		})
 	})
 }
 
@@ -431,32 +420,38 @@ func TestInboundValidity(t *testing.T) {
 			t.Fatalf("Error constructing ice.Agent")
 		}
 
-		a.handleInbound(buildMsg(stun.ClassRequest, a.localUfrag+":"+a.remoteUfrag, a.localPwd), local, remote)
-		if len(a.remoteCandidates) != 1 {
-			t.Fatal("Binding with valid values was unable to create prflx candidate")
-		}
-	})
+		err = a.run(func(a *Agent) {
+			a.selector = &controllingSelector{agent: a, log: a.log}
+			a.handleInbound(buildMsg(stun.ClassRequest, a.localUfrag+":"+a.remoteUfrag, a.localPwd), local, remote)
+			if len(a.remoteCandidates) != 1 {
+				t.Fatal("Binding with valid values was unable to create prflx candidate")
+			}
+		})
 
-	t.Run("Valid bind without fingerprint", func(t *testing.T) {
-		a, err := NewAgent(&AgentConfig{})
-		if err != nil {
-			t.Fatalf("Error constructing ice.Agent")
-		}
-
-		msg, err := stun.Build(stun.ClassRequest, stun.MethodBinding, stun.GenerateTransactionID(),
-			&stun.Username{Username: a.localUfrag + ":" + a.remoteUfrag},
-			&stun.MessageIntegrity{
-				Key: []byte(a.localPwd),
-			},
-		)
 		if err != nil {
-			t.Fatal(err)
+			t.Fatalf("Agent run failure: %v", err)
 		}
+	})
 
-		a.handleInbound(msg, local, remote)
-		if len(a.remoteCandidates) != 1 {
-			t.Fatal("Binding with valid values (but no fingerprint) was unable to create prflx candidate")
-		}
+	t.Run("Valid bind without fingerprint", func(t *testing.T) {
+		var config AgentConfig
+		runAgentTest(t, &config, func(a *Agent) {
+			a.selector = &controllingSelector{agent: a, log: a.log}
+			msg, err := stun.Build(stun.ClassRequest, stun.MethodBinding, stun.GenerateTransactionID(),
+				&stun.Username{Username: a.localUfrag + ":" + a.remoteUfrag},
+				&stun.MessageIntegrity{
+					Key: []byte(a.localPwd),
+				},
+			)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			a.handleInbound(msg, local, remote)
+			if len(a.remoteCandidates) != 1 {
+				t.Fatal("Binding with valid values (but no fingerprint) was unable to create prflx candidate")
+			}
+		})
 	})
 
 	t.Run("Success with invalid TransactionID", func(t *testing.T) {