Add Failed state to Agent

This is currently fatal because you can't do an restart.

Resolves #189

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 {

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())
+}