restore connectivity when UDPMuxDefault and NAT1To1IPs are used

PR #496 added a mechanism to close connections of duplicate candidates
and avoid memory leaks. When UDPMuxDefault and NAT1To1IPs are used
together, candidates may share the same connection, and candidates may
be equal, since their IPs are overridden by the one specified in
NAT1To1IPs. When a duplicate candidate is found, its connection is
closed, making impossible to use any other candidate that shares the
same connection.

This PR fixes the issue as it prevents duplicate candidates from
creating connections and getting their connection closed.

gather.go

@@ -255,13 +255,15 @@ func (a *Agent) gatherCandidatesLocalUDPMux(ctx context.Context) error { //nolin
  }
 
  localAddresses := a.udpMux.GetListenAddresses()
+ configs := make(map[CandidateHostConfig]struct{})
 
  for _, addr := range localAddresses {
  udpAddr, ok := addr.(*net.UDPAddr)
  if !ok {
  return errInvalidAddress
  }
  candidateIP := udpAddr.IP
+
  if a.extIPMapper != nil && a.extIPMapper.candidateType == CandidateTypeHost {
  mappedIP, err := a.extIPMapper.findExternalIP(candidateIP.String())
  if err != nil {
@@ -271,18 +273,27 @@ func (a *Agent) gatherCandidatesLocalUDPMux(ctx context.Context) error { //nolin
 
  candidateIP = mappedIP
  }
-
- conn, err := a.udpMux.GetConn(a.localUfrag, udpAddr)
- if err != nil {
- return err
- }
  hostConfig := CandidateHostConfig{
  Network: udp,
  Address: candidateIP.String(),
  Port: udpAddr.Port,
  Component: ComponentRTP,
  }
 
+ // detect a duplicate candidate before calling addCandidate().
+ // otherwise, addCandidate() detects the duplicate candidate
+ // and close its connection, invalidating all candidates
+ // that share the same connection.
+ if _, ok := configs[hostConfig]; ok {
+ continue
+ }
+ configs[hostConfig] = struct{}{}
+
+ conn, err := a.udpMux.GetConn(a.localUfrag, udpAddr)
+ if err != nil {
+ return err
+ }
+
  c, err := NewCandidateHost(&hostConfig)
  if err != nil {
  closeConnAndLog(conn, a.log, fmt.Sprintf("Failed to create host mux candidate: %s %d: %v", candidateIP, udpAddr.Port, err))

gather_test.go

@@ -599,6 +599,47 @@ func TestTURNProxyDialer(t *testing.T) {
  assert.NoError(t, a.Close())
 }
 
+// Assert that candidates are given and connections are valid
+// when using UDPMuxDefault and NAT1To1IPs.
+func TestUDPMuxDefaultWithNAT1To1IPsUsage(t *testing.T) {
+ report := test.CheckRoutines(t)
+ defer report()
+
+ lim := test.TimeOut(time.Second * 30)
+ defer lim.Stop()
+
+ conn, err := net.ListenPacket("udp4", ":0")
+ assert.NoError(t, err)
+ defer conn.Close()
+
+ mux := NewUDPMuxDefault(UDPMuxParams{
+ UDPConn: conn,
+ })
+ defer mux.Close()
+
+ a, err := NewAgent(&AgentConfig{
+ NAT1To1IPs: []string{"1.2.3.4"},
+ NAT1To1IPCandidateType: CandidateTypeHost,
+ UDPMux: mux,
+ })
+ assert.NoError(t, err)
+
+ gatherCandidateDone := make(chan struct{})
+ assert.NoError(t, a.OnCandidate(func(c Candidate) {
+ if c == nil {
+ close(gatherCandidateDone)
+ } else {
+ assert.Equal(t, "1.2.3.4", c.Address())
+ }
+ }))
+ assert.NoError(t, a.GatherCandidates())
+ <-gatherCandidateDone
+
+ assert.NotEqual(t, 0, len(mux.connsIPv4))
+
+ assert.NoError(t, a.Close())
+}
+
 // Assert that candidates are given for each mux in a MultiUDPMux
 func TestMultiUDPMuxUsage(t *testing.T) {
  report := test.CheckRoutines(t)