Repository: pions/webrtc
Pull request #325 (2018-12-28 15:14:38)

Add settingEngine hooks to modify ICE timeouts
Relates to #324

coveralls: ## Pull Request Test Coverage Report for [Build 1567](https://coveralls.io/builds/20827125)

* **18** of **33**   **(54.55%)**  changed or added relevant lines in **4** files are covered.
* **7** unchanged lines in **3** files lost coverage.
* Overall coverage increased (+**0.08%**) to **72.873%**

---

|  Changes Missing Coverage | Covered Lines | Changed/Added Lines | % |
| :-----|--------------|--------|---: |
| [settingengine.go](https://coveralls.io/builds/20827125/source?filename=settingengine.go#L15) | 0 | 15 | 0.0%
<!-- | **Total:** | **18** | **33** | **54.55%** | -->

|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [internal/mux/mux.go](https://coveralls.io/builds/20827125/source?filename=internal%2Fmux%2Fmux.go#L109) | 1 | 90.41% |
| [pkg/ice/transport.go](https://coveralls.io/builds/20827125/source?filename=pkg%2Fice%2Ftransport.go#L61) | 2 | 54.29% |
| [pkg/ice/agent.go](https://coveralls.io/builds/20827125/source?filename=pkg%2Fice%2Fagent.go#L83) | 4 | 82.02% |
<!-- | **Total:** | **7** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/20827125/badge)](https://coveralls.io/builds/20827125) |
| :-- | --: |
| Change from base [Build 1558](https://coveralls.io/builds/20823453): |  0.08% |
| Covered Lines: | 4505 |
| Relevant Lines: | 6182 |

---
##### 💛  - [Coveralls](https://coveralls.io)


Patch:
diff --git a/internal/network/manager.go b/internal/network/manager.go
index bb9c44a6..788c1b6b 100644
--- a/internal/network/manager.go
+++ b/internal/network/manager.go
@@ -45,8 +45,7 @@ type Manager struct {
 }
 
 // NewManager creates a new network.Manager
-func NewManager(urls []*ice.URL, ntf ICENotifier, minport, maxport uint16) (*Manager, error) {
-	config := &ice.AgentConfig{Urls: urls, Notifier: ntf, PortMin: minport, PortMax: maxport}
+func NewManager(config *ice.AgentConfig) (*Manager, error) {
 	iceAgent, err := ice.NewAgent(config)
 
 	if err != nil {
diff --git a/pkg/ice/agent.go b/pkg/ice/agent.go
index 00d47684..c650bfb9 100644
--- a/pkg/ice/agent.go
+++ b/pkg/ice/agent.go
@@ -19,10 +19,10 @@ const (
 	taskLoopInterval = 2 * time.Second
 
 	// keepaliveInterval used to keep candidates alive
-	keepaliveInterval = 10 * time.Second
+	defaultKeepaliveInterval = 10 * time.Second
 
-	// connectionTimeout used to declare a connection dead
-	connectionTimeout = 30 * time.Second
+	// defaultConnectionTimeout used to declare a connection dead
+	defaultConnectionTimeout = 30 * time.Second
 )
 
 // Agent represents the ICE agent
@@ -50,6 +50,14 @@ type Agent struct {
 	portmin uint16
 	portmax uint16
 
+	//How long should a pair stay quiet before we declare it dead?
+	//0 means never timeout
+	connectionTimeout time.Duration
+
+	//How often should we send keepalive packets?
+	//0 means never
+	keepaliveInterval time.Duration
+
 	localUfrag      string
 	localPwd        string
 	localCandidates map[NetworkType][]*Candidate
@@ -100,6 +108,15 @@ type AgentConfig struct {
 	// PortMin and PortMax are optional. Leave them 0 for the default UDP port allocation strategy.
 	PortMin uint16
 	PortMax uint16
+
+	// ConnectionTimeout defaults to 30 seconds when this property is nil.
+	// If the duration is 0, we will never timeout this connection.
+	ConnectionTimeout *time.Duration
+	// KeepaliveInterval determines how often should we send ICE
+	// keepalives (should be less then connectiontimeout above)
+	// when this is nil, it defaults to 10 seconds.
+	// A keepalive interval of 0 means we never send keepalive packets
+	KeepaliveInterval *time.Duration
 }
 
 // NewAgent creates a new Agent
@@ -126,6 +143,19 @@ func NewAgent(config *AgentConfig) (*Agent, error) {
 		portmax:     config.PortMax,
 	}
 
+	// connectionTimeout used to declare a connection dead
+	if config.ConnectionTimeout == nil {
+		a.connectionTimeout = defaultConnectionTimeout
+	} else {
+		a.connectionTimeout = *config.ConnectionTimeout
+	}
+
+	if config.KeepaliveInterval == nil {
+		a.keepaliveInterval = defaultKeepaliveInterval
+	} else {
+		a.keepaliveInterval = *config.KeepaliveInterval
+	}
+
 	// Initialize local candidates
 	a.gatherCandidatesLocal()
 	a.gatherCandidatesReflective(config.Urls)
@@ -401,7 +431,8 @@ func (a *Agent) validateSelectedPair() bool {
 		return false
 	}
 
-	if time.Since(a.selectedPair.remote.LastReceived()) > connectionTimeout {
+	if (a.connectionTimeout != 0) &&
+		(time.Since(a.selectedPair.remote.LastReceived()) > a.connectionTimeout) {
 		a.selectedPair = nil
 		a.updateConnectionState(ConnectionStateDisconnected)
 		return false
@@ -418,7 +449,8 @@ func (a *Agent) checkKeepalive() {
 		return
 	}
 
-	if time.Since(a.selectedPair.local.LastSent()) > keepaliveInterval {
+	if (a.keepaliveInterval != 0) &&
+		(time.Since(a.selectedPair.local.LastSent()) > a.keepaliveInterval) {
 		a.keepaliveCandidate(a.selectedPair.local, a.selectedPair.remote)
 	}
 }
diff --git a/rtcpeerconnection.go b/rtcpeerconnection.go
index 6dd54872..dcc53423 100644
--- a/rtcpeerconnection.go
+++ b/rtcpeerconnection.go
@@ -160,7 +160,12 @@ func New(configuration RTCConfiguration) (*RTCPeerConnection, error) {
 		}
 	}
 
-	pc.networkManager, err = network.NewManager(urls, pc.iceStateChange, defaultSettingEngine.EphemeralUDP.PortMin, defaultSettingEngine.EphemeralUDP.PortMax)
+	pc.networkManager, err = network.NewManager(
+		&ice.AgentConfig{Urls: urls, Notifier: pc.iceStateChange,
+			PortMin:           defaultSettingEngine.EphemeralUDP.PortMin,
+			PortMax:           defaultSettingEngine.EphemeralUDP.PortMax,
+			ConnectionTimeout: defaultSettingEngine.Timeout.ICEConnection,
+			KeepaliveInterval: defaultSettingEngine.Timeout.ICEKeepalive})
 	if err != nil {
 		return nil, err
 	}
diff --git a/settingengine.go b/settingengine.go
index 98422fe6..a15db0b9 100644
--- a/settingengine.go
+++ b/settingengine.go
@@ -1,6 +1,10 @@
 package webrtc
 
-import "github.com/pions/webrtc/pkg/ice"
+import (
+	"time"
+
+	"github.com/pions/webrtc/pkg/ice"
+)
 
 var defaultSettingEngine = newSettingEngine()
 
@@ -8,13 +12,7 @@ var defaultSettingEngine = newSettingEngine()
 // ICE UDP connections can allocate from. This setting currently only
 // affects host candidates, not server reflexive candidates.
 func SetEphemeralUDPPortRange(portMin, portMax uint16) error {
-	if portMax < portMin {
-		return ice.ErrPort
-	}
-
-	defaultSettingEngine.EphemeralUDP.PortMin = portMin
-	defaultSettingEngine.EphemeralUDP.PortMax = portMax
-	return nil
+	return defaultSettingEngine.SetEphemeralUDPPortRange(portMin, portMax)
 }
 
 // DetachDataChannels enables detaching data channels. When enabled
@@ -24,6 +22,12 @@ func DetachDataChannels() {
 	defaultSettingEngine.DetachDataChannels()
 }
 
+// SetConnectionTimeout sets the amount of silence needed on a given candidate pair
+// before the ICE agent considers the pair timed out.
+func SetConnectionTimeout(connectionTimeout, keepAlive time.Duration) {
+	defaultSettingEngine.SetConnectionTimeout(connectionTimeout, keepAlive)
+}
+
 // settingEngine allows influencing behavior in ways that are not
 // supported by the WebRTC API. This allows us to support additional
 // use-cases without deviating from the WebRTC API elsewhere.
@@ -35,6 +39,10 @@ type settingEngine struct {
 	Detach struct {
 		DataChannels bool
 	}
+	Timeout struct {
+		ICEConnection *time.Duration
+		ICEKeepalive  *time.Duration
+	}
 }
 
 // DetachDataChannels enables detaching data channels. When enabled
@@ -44,6 +52,26 @@ func (e *settingEngine) DetachDataChannels() {
 	e.Detach.DataChannels = true
 }
 
+// SetConnectionTimeout sets the amount of silence needed on a given candidate pair
+// before the ICE agent considers the pair timed out.
+func (e *settingEngine) SetConnectionTimeout(connectionTimeout, keepAlive time.Duration) {
+	e.Timeout.ICEConnection = &connectionTimeout
+	e.Timeout.ICEKeepalive = &keepAlive
+}
+
+// SetEphemeralUDPPortRange limits the pool of ephemeral ports that
+// ICE UDP connections can allocate from. This setting currently only
+// affects host candidates, not server reflexive candidates.
+func (e *settingEngine) SetEphemeralUDPPortRange(portMin, portMax uint16) error {
+	if portMax < portMin {
+		return ice.ErrPort
+	}
+
+	e.EphemeralUDP.PortMin = portMin
+	e.EphemeralUDP.PortMax = portMax
+	return nil
+}
+
 func newSettingEngine() *settingEngine {
 	return new(settingEngine)
 }


Test patch:
diff --git a/internal/mux/mux_test.go b/internal/mux/mux_test.go
index da43d564..fc808618 100644
--- a/internal/mux/mux_test.go
+++ b/internal/mux/mux_test.go
@@ -65,3 +65,23 @@ func pipeMemory() (*Endpoint, net.Conn, func(*testing.T)) {
 
 	return e, cb, stop
 }
+
+func TestNoEndpoints(t *testing.T) {
+	// In memory pipe
+	ca, cb := net.Pipe()
+	err := cb.Close()
+	if err != nil {
+		panic("Failed to close network pipe")
+	}
+	m := NewMux(ca, 8192)
+	m.dispatch(make([]byte, 1))
+	err = m.Close()
+	if err != nil {
+		t.Fatalf("Failed to close empty mux")
+	}
+	err = ca.Close()
+	if err != nil {
+		panic("Failed to close network pipe")
+	}
+
+}
diff --git a/pkg/ice/transport_test.go b/pkg/ice/transport_test.go
index ccbbbd71..dde09b85 100644
--- a/pkg/ice/transport_test.go
+++ b/pkg/ice/transport_test.go
@@ -21,6 +21,84 @@ func TestStressDuplex(t *testing.T) {
 	stressDuplex(t)
 }
 
+func testTimeout(t *testing.T, c *Conn, timeout time.Duration) {
+	const pollrate = 100 * time.Millisecond
+	statechan := make(chan ConnectionState)
+	ticker := time.NewTicker(pollrate)
+
+	for cnt := time.Duration(0); cnt <= timeout+taskLoopInterval; cnt += pollrate {
+		<-ticker.C
+		err := c.agent.run(func(agent *Agent) {
+			statechan <- agent.connectionState
+		})
+
+		if err != nil {
+			//we should never get here.
+			panic(err)
+		}
+
+		cs := <-statechan
+		if cs != ConnectionStateConnected {
+			if cnt < timeout {
+				t.Fatalf("Connection timed out early. (after %d ms)", cnt/time.Millisecond)
+			} else {
+				return
+			}
+		}
+	}
+	t.Fatalf("Connection failed to time out in time.")
+
+}
+
+func TestTimeout(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping test in short mode.")
+	}
+
+	ca, cb := pipe()
+	err := cb.Close()
+
+	if err != nil {
+		//we should never get here.
+		panic(err)
+	}
+
+	testTimeout(t, ca, 30*time.Second)
+
+	ca, cb = pipeWithTimeout(5*time.Second, 3*time.Second)
+	err = cb.Close()
+
+	if err != nil {
+		//we should never get here.
+		panic(err)
+	}
+
+	testTimeout(t, ca, 5*time.Second)
+}
+
+func TestReadClosed(t *testing.T) {
+	ca, cb := pipe()
+
+	err := ca.Close()
+	if err != nil {
+		//we should never get here.
+		panic(err)
+	}
+
+	err = cb.Close()
+	if err != nil {
+		//we should never get here.
+		panic(err)
+	}
+
+	empty := make([]byte, 10)
+	_, err = ca.Read(empty)
+	if err == nil {
+		t.Fatalf("Reading from a closed channel should return an error")
+	}
+
+}
+
 func stressDuplex(t *testing.T) {
 	ca, cb := pipe()
 
@@ -72,26 +150,7 @@ func check(err error) {
 	}
 }
 
-func pipe() (*Conn, *Conn) {
-	var urls []*URL
-
-	aNotifier, aConnected := onConnected()
-	bNotifier, bConnected := onConnected()
-
-	aAgent, err := NewAgent(&AgentConfig{Urls: urls, Notifier: aNotifier})
-
-	if err != nil {
-		//we should never get here.
-		panic(err)
-	}
-
-	bAgent, err := NewAgent(&AgentConfig{Urls: urls, Notifier: bNotifier})
-
-	if err != nil {
-		//we should never get here.
-		panic(err)
-	}
-
+func connect(aAgent, bAgent *Agent) (*Conn, *Conn) {
 	// Manual signaling
 	aUfrag, aPwd := aAgent.GetLocalUserCredentials()
 	bUfrag, bPwd := bAgent.GetLocalUserCredentials()
@@ -123,6 +182,60 @@ func pipe() (*Conn, *Conn) {
 
 	// Ensure accepted
 	<-accepted
+	return aConn, bConn
+}
+
+func pipe() (*Conn, *Conn) {
+	var urls []*URL
+
+	aNotifier, aConnected := onConnected()
+	bNotifier, bConnected := onConnected()
+
+	aAgent, err := NewAgent(&AgentConfig{Urls: urls, Notifier: aNotifier})
+
+	if err != nil {
+		//we should never get here.
+		panic(err)
+	}
+
+	bAgent, err := NewAgent(&AgentConfig{Urls: urls, Notifier: bNotifier})
+
+	if err != nil {
+		//we should never get here.
+		panic(err)
+	}
+
+	aConn, bConn := connect(aAgent, bAgent)
+
+	// Ensure pair selected
+	// Note: this assumes ConnectionStateConnected is thrown after selecting the final pair
+	<-aConnected
+	<-bConnected
+
+	return aConn, bConn
+}
+
+func pipeWithTimeout(ICETimeout time.Duration, ICEKeepalive time.Duration) (*Conn, *Conn) {
+	var urls []*URL
+
+	aNotifier, aConnected := onConnected()
+	bNotifier, bConnected := onConnected()
+
+	aAgent, err := NewAgent(&AgentConfig{Urls: urls, Notifier: aNotifier, ConnectionTimeout: &ICETimeout, KeepaliveInterval: &ICEKeepalive})
+
+	if err != nil {
+		//we should never get here.
+		panic(err)
+	}
+
+	bAgent, err := NewAgent(&AgentConfig{Urls: urls, Notifier: bNotifier, ConnectionTimeout: &ICETimeout, KeepaliveInterval: &ICEKeepalive})
+
+	if err != nil {
+		//we should never get here.
+		panic(err)
+	}
+
+	aConn, bConn := connect(aAgent, bAgent)
 
 	// Ensure pair selected
 	// Note: this assumes ConnectionStateConnected is thrown after selecting the final pair