diff --git a/agent.go b/agent.go index e554252..72bbbb0 100644 --- a/agent.go +++ b/agent.go @@ -380,32 +380,63 @@ func (a *Agent) startConnectivityChecks(isControlling bool, remoteUfrag, remoteP // TODO this should be dynamic, and grow when the connection is stable a.requestConnectivityCheck() agent.connectivityTicker = time.NewTicker(a.taskLoopInterval) + go a.connectivityChecks() + }) +} - go func() { - contact := func() { - if err := a.run(func(a *Agent) { - a.selector.ContactCandidates() - }); err != nil { - a.log.Warnf("taskLoop failed: %v", err) +func (a *Agent) connectivityChecks() { + lastConnectionState := ConnectionState(0) + checkingDuration := time.Time{} + + contact := func() { + if err := a.run(func(a *Agent) { + defer func() { + lastConnectionState = a.connectionState + }() + + switch a.connectionState { + case ConnectionStateFailed: + // The connection is currently failed so don't send any checks + // In the future it may be restarted though + return + case ConnectionStateChecking: + // We have just entered checking for the first time so update our checking timer + if lastConnectionState != a.connectionState { + checkingDuration = time.Now() } - } - for { - select { - case <-a.forceCandidateContact: - contact() - case <-a.connectivityTicker.C: - contact() - case <-a.done: + // We have been in checking longer then Disconnect+Failed timeout, set the connection to Failed + if time.Since(checkingDuration) > a.disconnectTimeout+a.failedTimeout { + a.updateConnectionState(ConnectionStateFailed) return } } - }() - }) + + a.selector.ContactCandidates() + }); err != nil { + a.log.Warnf("taskLoop failed: %v", err) + } + } + + for { + select { + case <-a.forceCandidateContact: + contact() + case <-a.connectivityTicker.C: + contact() + case <-a.done: + return + } + } } func (a *Agent) updateConnectionState(newState ConnectionState) { if a.connectionState != newState { + // Connection has gone to failed, release all gathered candidates + if newState == ConnectionStateFailed { + a.deleteAllCandidates() + } + a.log.Infof("Setting new connection state: %s", newState) a.connectionState = newState @@ -527,7 +558,6 @@ func (a *Agent) validateSelectedPair() bool { switch { case totalTimeToFailure != 0 && disconnectedTime > totalTimeToFailure: - a.deleteAllCandidates() a.updateConnectionState(ConnectionStateFailed) case a.disconnectTimeout != 0 && disconnectedTime > a.disconnectTimeout: a.updateConnectionState(ConnectionStateDisconnected) diff --git a/agent_test.go b/agent_test.go index e57be05..9bf493a 100644 --- a/agent_test.go +++ b/agent_test.go @@ -357,20 +357,24 @@ func TestConnectivityOnStartup(t *testing.T) { report := test.CheckRoutines(t) defer report() - stunServerURL := &URL{ - Scheme: SchemeTypeSTUN, - Host: "1.2.3.4", - Port: 3478, - Proto: ProtoTypeUDP, - } + // Create a network with two interfaces + wan, err := vnet.NewRouter(&vnet.RouterConfig{ + CIDR: "0.0.0.0/0", + LoggerFactory: logging.NewDefaultLoggerFactory(), + }) + assert.NoError(t, err) - natType := &vnet.NATType{ - MappingBehavior: vnet.EndpointIndependent, - FilteringBehavior: vnet.EndpointIndependent, - } - v, err := buildVNet(natType, natType) - require.NoError(t, err, "should succeed") - defer v.close() + net0 := vnet.NewNet(&vnet.NetConfig{ + StaticIPs: []string{"192.168.0.1"}, + }) + assert.NoError(t, wan.AddNet(net0)) + + net1 := vnet.NewNet(&vnet.NetConfig{ + StaticIPs: []string{"192.168.0.2"}, + }) + assert.NoError(t, wan.AddNet(net1)) + + assert.NoError(t, wan.Start()) aNotifier, aConnected := onConnected() bNotifier, bConnected := onConnected() @@ -379,11 +383,10 @@ func TestConnectivityOnStartup(t *testing.T) { wg.Add(2) cfg0 := &AgentConfig{ - Urls: []*URL{stunServerURL}, Trickle: true, NetworkTypes: supportedNetworkTypes, MulticastDNSMode: MulticastDNSModeDisabled, - Net: v.net0, + Net: net0, taskLoopInterval: time.Hour, } @@ -399,11 +402,10 @@ func TestConnectivityOnStartup(t *testing.T) { require.NoError(t, aAgent.GatherCandidates()) cfg1 := &AgentConfig{ - Urls: []*URL{stunServerURL}, Trickle: true, NetworkTypes: supportedNetworkTypes, MulticastDNSMode: MulticastDNSModeDisabled, - Net: v.net1, + Net: net1, taskLoopInterval: time.Hour, } @@ -425,6 +427,7 @@ func TestConnectivityOnStartup(t *testing.T) { <-aConnected <-bConnected + assert.NoError(t, wan.Stop()) if !closePipe(t, aConn, bConn) { return } @@ -1269,9 +1272,9 @@ func TestAgentCredentials(t *testing.T) { assert.EqualError(t, err, ErrLocalPwdInsufficientBits.Error()) } -// Assert that Agent on Failure flushes all existing candidates +// Assert that Agent on Failure deletes all existing candidates // User can then do an ICE Restart to bring agent back -func TestConnectionStateFailedFlushCandidates(t *testing.T) { +func TestConnectionStateFailedDeleteAllCandidates(t *testing.T) { lim := test.TimeOut(time.Second * 5) defer lim.Stop() @@ -1282,7 +1285,6 @@ func TestConnectionStateFailedFlushCandidates(t *testing.T) { KeepaliveInterval := time.Duration(0) cfg := &AgentConfig{ - Urls: []*URL{}, NetworkTypes: supportedNetworkTypes, DisconnectTimeout: &oneSecond, FailedTimeout: &oneSecond, @@ -1297,18 +1299,84 @@ func TestConnectionStateFailedFlushCandidates(t *testing.T) { assert.NoError(t, err) isFailed := make(chan interface{}) - err = aAgent.OnConnectionStateChange(func(c ConnectionState) { + assert.NoError(t, aAgent.OnConnectionStateChange(func(c ConnectionState) { if c == ConnectionStateFailed { close(isFailed) } - }) - assert.NoError(t, err) + })) connect(aAgent, bAgent) <-isFailed - assert.Equal(t, len(aAgent.remoteCandidates), 0) - assert.Equal(t, len(bAgent.localCandidates), 0) + done := make(chan struct{}) + assert.NoError(t, aAgent.run(func(agent *Agent) { + assert.Equal(t, len(aAgent.remoteCandidates), 0) + assert.Equal(t, len(aAgent.localCandidates), 0) + close(done) + })) + <-done + + assert.NoError(t, aAgent.Close()) + assert.NoError(t, bAgent.Close()) +} + +// Assert that the ICE Agent can go directly from Connecting -> Failed on both sides +func TestConnectionStateConnectingToFailed(t *testing.T) { + lim := test.TimeOut(time.Second * 5) + defer lim.Stop() + + report := test.CheckRoutines(t) + defer report() + + oneSecond := time.Second + KeepaliveInterval := time.Duration(0) + + cfg := &AgentConfig{ + DisconnectTimeout: &oneSecond, + FailedTimeout: &oneSecond, + KeepaliveInterval: &KeepaliveInterval, + taskLoopInterval: 250 * time.Millisecond, + } + + aAgent, err := NewAgent(cfg) + assert.NoError(t, err) + + bAgent, err := NewAgent(cfg) + assert.NoError(t, err) + + var isFailed sync.WaitGroup + var isChecking sync.WaitGroup + + isFailed.Add(2) + isChecking.Add(2) + + connectionStateCheck := func(c ConnectionState) { + switch c { + case ConnectionStateFailed: + isFailed.Done() + case ConnectionStateChecking: + isChecking.Done() + case ConnectionStateConnected: + case ConnectionStateCompleted: + t.Errorf("Unexpected ConnectionState: %v", c) + } + } + + assert.NoError(t, aAgent.OnConnectionStateChange(connectionStateCheck)) + assert.NoError(t, bAgent.OnConnectionStateChange(connectionStateCheck)) + + go func() { + _, err := aAgent.Accept(context.TODO(), "InvalidFrag", "InvalidPwd") + assert.Error(t, err) + }() + + go func() { + _, err := bAgent.Dial(context.TODO(), "InvalidFrag", "InvalidPwd") + assert.Error(t, err) + }() + + isChecking.Wait() + isFailed.Wait() assert.NoError(t, aAgent.Close()) assert.NoError(t, bAgent.Close()) diff --git a/selection.go b/selection.go index 45a6e4e..f8845a8 100644 --- a/selection.go +++ b/selection.go @@ -26,31 +26,6 @@ type controllingSelector struct { func (s *controllingSelector) Start() { s.startTime = time.Now() - go func() { - select { - case <-s.agent.done: - return - case <-time.After(s.agent.candidateSelectionTimeout): - } - - err := s.agent.run(func(a *Agent) { - if s.nominatedPair == nil { - p := s.agent.getBestValidCandidatePair() - if p == nil { - s.log.Trace("check timeout reached and no valid candidate pair found, marking connection as failed") - s.agent.updateConnectionState(ConnectionStateFailed) - } else { - s.log.Tracef("check timeout reached, nominating (%s, %s)", p.local.String(), p.remote.String()) - s.nominatedPair = p - s.nominatePair(p) - } - } - }) - - if err != nil { - s.log.Errorf("error processing checkCandidatesTimeout handler %v", err.Error()) - } - }() } func (s *controllingSelector) isNominatable(c Candidate) bool { @@ -193,13 +168,11 @@ func (s *controllingSelector) PingCandidate(local, remote Candidate) { } type controlledSelector struct { - startTime time.Time - agent *Agent - log logging.LeveledLogger + agent *Agent + log logging.LeveledLogger } func (s *controlledSelector) Start() { - s.startTime = time.Now() } func (s *controlledSelector) ContactCandidates() { @@ -209,12 +182,7 @@ func (s *controlledSelector) ContactCandidates() { s.agent.checkKeepalive() } } else { - if time.Since(s.startTime) > s.agent.candidateSelectionTimeout { - s.log.Trace("check timeout reached and no valid candidate pair found, marking connection as failed") - s.agent.updateConnectionState(ConnectionStateFailed) - } else { - s.agent.pingAllCandidates() - } + s.agent.pingAllCandidates() } }