diff --git a/agent.go b/agent.go index 8fd7bf7..e7a16cc 100644 --- a/agent.go +++ b/agent.go @@ -121,7 +121,7 @@ type Agent struct { // How long connectivity checks can fail before the ICE Agent // goes to failed - failedTimeout time.Duration //nolint + failedTimeout time.Duration // How often should we send keepalive packets? // 0 means never @@ -764,9 +764,21 @@ func (a *Agent) validateSelectedPair() bool { return false } - if (a.disconnectTimeout != 0) && (time.Since(selectedPair.remote.LastReceived()) > a.disconnectTimeout) { + disconnectedTime := time.Since(selectedPair.remote.LastReceived()) + + // Only allow transitions to failed if a.failedTimeout is non-zero + totalTimeToFailure := a.failedTimeout + if totalTimeToFailure != 0 { + totalTimeToFailure += a.disconnectTimeout + } + + switch { + case totalTimeToFailure != 0 && disconnectedTime > totalTimeToFailure: + a.deleteAllCandidates() + a.updateConnectionState(ConnectionStateFailed) + case a.disconnectTimeout != 0 && disconnectedTime > a.disconnectTimeout: a.updateConnectionState(ConnectionStateDisconnected) - } else { + default: a.updateConnectionState(ConnectionStateConnected) } @@ -940,25 +952,8 @@ func (a *Agent) Close() error { agent.err.Store(ErrClosed) close(agent.done) - // Cleanup all candidates - for net, cs := range agent.localCandidates { - for _, c := range cs { - err := c.close() - if err != nil { - a.log.Warnf("Failed to close candidate %s: %v", c, err) - } - } - delete(agent.localCandidates, net) - } - for net, cs := range agent.remoteCandidates { - for _, c := range cs { - err := c.close() - if err != nil { - a.log.Warnf("Failed to close candidate %s: %v", c, err) - } - } - delete(agent.remoteCandidates, net) - } + a.deleteAllCandidates() + if err := a.buffer.Close(); err != nil { a.log.Warnf("failed to close buffer: %v", err) } @@ -978,6 +973,29 @@ func (a *Agent) Close() error { return nil } +// Remove all candidates. This closes any listening sockets +// and removes both the local and remote candidate lists. +// +// This is used for restarts, failures and on close +func (a *Agent) deleteAllCandidates() { + for net, cs := range a.localCandidates { + for _, c := range cs { + if err := c.close(); err != nil { + a.log.Warnf("Failed to close candidate %s: %v", c, err) + } + } + delete(a.localCandidates, net) + } + for net, cs := range a.remoteCandidates { + for _, c := range cs { + if err := c.close(); err != nil { + a.log.Warnf("Failed to close candidate %s: %v", c, err) + } + } + delete(a.remoteCandidates, net) + } +} + func (a *Agent) findRemoteCandidate(networkType NetworkType, addr net.Addr) Candidate { ip, port, err := addrIPAndPort(addr) if err != nil { diff --git a/agent_test.go b/agent_test.go index a691aba..e57be05 100644 --- a/agent_test.go +++ b/agent_test.go @@ -691,7 +691,7 @@ func TestInvalidAgentStarts(t *testing.T) { assert.NoError(t, a.Close()) } -// Assert that Agent emits Connecting/Connected/Disconnected/Closed messages +// Assert that Agent emits Connecting/Connected/Disconnected/Failed/Closed messages func TestConnectionStateCallback(t *testing.T) { lim := test.TimeOut(time.Second * 5) defer lim.Stop() @@ -703,12 +703,15 @@ func TestConnectionStateCallback(t *testing.T) { wg.Add(2) disconnectDuration := time.Second + failedDuration := time.Second KeepaliveInterval := time.Duration(0) + cfg := &AgentConfig{ Urls: []*URL{}, Trickle: true, NetworkTypes: supportedNetworkTypes, DisconnectTimeout: &disconnectDuration, + FailedTimeout: &failedDuration, KeepaliveInterval: &KeepaliveInterval, taskLoopInterval: 500 * time.Millisecond, } @@ -750,6 +753,7 @@ func TestConnectionStateCallback(t *testing.T) { isChecking := make(chan interface{}) isConnected := make(chan interface{}) isDisconnected := make(chan interface{}) + isFailed := make(chan interface{}) isClosed := make(chan interface{}) err = aAgent.OnConnectionStateChange(func(c ConnectionState) { switch c { @@ -759,6 +763,8 @@ func TestConnectionStateCallback(t *testing.T) { close(isConnected) case ConnectionStateDisconnected: close(isDisconnected) + case ConnectionStateFailed: + close(isFailed) case ConnectionStateClosed: close(isClosed) } @@ -773,6 +779,7 @@ func TestConnectionStateCallback(t *testing.T) { <-isChecking <-isConnected <-isDisconnected + <-isFailed assert.NoError(t, aAgent.Close()) assert.NoError(t, bAgent.Close()) @@ -1261,3 +1268,48 @@ func TestAgentCredentials(t *testing.T) { _, err = NewAgent(&AgentConfig{Trickle: true, LocalPwd: "xxxxxx", LoggerFactory: log}) assert.EqualError(t, err, ErrLocalPwdInsufficientBits.Error()) } + +// Assert that Agent on Failure flushes all existing candidates +// User can then do an ICE Restart to bring agent back +func TestConnectionStateFailedFlushCandidates(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{ + Urls: []*URL{}, + NetworkTypes: supportedNetworkTypes, + 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) + + isFailed := make(chan interface{}) + err = 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) + + assert.NoError(t, aAgent.Close()) + assert.NoError(t, bAgent.Close()) +}