zishuo/ice
1
0
Fork 0
mirror of https://github.com/pion/ice.git synced 2025-10-23 23:43:57 +08:00
Code Issues Packages Projects Releases Wiki Activity

Revert active TCP candidate support

Browse Source 
This reverts commit 00bbd2954c
and 1d502ca6ec
This commit is contained in:
Sean DuBois
2023-05-24 00:05:31 -04:00
parent 7715befedc
commit abc1ca33b5
10 changed files with 78 additions and 390 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
agent.go
View File

@@ -9,7 +9,6 @@ import (
"context"
"fmt"
"net"
"strconv"
"strings"
"sync"
"sync/atomic"
@@ -73,8 +72,6 @@ type Agent struct {
prflxAcceptanceMinWait time.Duration
relayAcceptanceMinWait time.Duration
tcpPriorityOffset uint16
portMin uint16
portMax uint16
@@ -588,44 +585,6 @@ func (a *Agent) getBestValidCandidatePair() *CandidatePair {
}
func (a *Agent) addPair(local, remote Candidate) *CandidatePair {
if local.TCPType() == TCPTypeActive && remote.TCPType() == TCPTypeActive {
return nil
}
if local.TCPType() == TCPTypeActive && remote.TCPType() == TCPTypePassive {
addressToConnect := net.JoinHostPort(remote.Address(), strconv.Itoa(remote.Port()))
conn, err := a.net.Dial("tcp", addressToConnect)
if err != nil {
a.log.Errorf("Failed to dial TCP address %s: %v", addressToConnect, err)
return nil
}
packetConn := newTCPPacketConn(tcpPacketParams{
ReadBuffer: tcpReadBufferSize,
LocalAddr: conn.LocalAddr(),
Logger: a.log,
})
if err = packetConn.AddConn(conn, nil); err != nil {
a.log.Errorf("Failed to add TCP connection: %v", err)
return nil
}
localAddress, ok := conn.LocalAddr().(*net.TCPAddr)
if !ok {
a.log.Errorf("Failed to cast local address to TCP address")
return nil
}
localCandidateHost, ok := local.(*CandidateHost)
if !ok {
a.log.Errorf("Failed to cast local candidate to CandidateHost")
return nil
}
localCandidateHost.port = localAddress.Port // This causes a data race with candidateBase.Port()
local.start(a, packetConn, a.startedCh)
}
p := newCandidatePair(local, remote, a.isControlling)
a.checklist = append(a.checklist, p)
return p
@@ -802,9 +761,7 @@ func (a *Agent) addCandidate(ctx context.Context, c Candidate, candidateConn net
}
}
if c.TCPType() != TCPTypeActive {
c.start(a, candidateConn, a.startedCh)
}
c.start(a, candidateConn, a.startedCh)
set = append(set, c)
a.localCandidates[c.NetworkType()] = set
@@ -1072,11 +1029,6 @@ func (a *Agent) handleInbound(m *stun.Message, local Candidate, remote net.Addr)
return
}
remoteTCPType := TCPTypeUnspecified
if local.TCPType() == TCPTypePassive {
remoteTCPType = TCPTypeActive
}
prflxCandidateConfig := CandidatePeerReflexiveConfig{
Network: networkType.String(),
Address: ip.String(),
@@ -1084,7 +1036,6 @@ func (a *Agent) handleInbound(m *stun.Message, local Candidate, remote net.Addr)
Component: local.Component(),
RelAddr: "",
RelPort: 0,
TCPType: remoteTCPType,
}
prflxCandidate, err := NewCandidatePeerReflexive(&prflxCandidateConfig)

162
agent_active_tcp_test.go
View File

@@ -1,162 +0,0 @@
// SPDX-FileCopyrightText: 2023 The Pion community <https://pion.ly>
// SPDX-License-Identifier: MIT
//go:build !js
// +build !js
package ice
import (
"net"
"testing"
"time"
"github.com/pion/logging"
"github.com/pion/transport/v2/stdnet"
"github.com/pion/transport/v2/test"
"github.com/stretchr/testify/require"
)
func getLocalIPAddress(t *testing.T, networkType NetworkType) net.IP {
net, err := stdnet.NewNet()
require.NoError(t, err)
localIPs, err := localInterfaces(net, nil, nil, []NetworkType{networkType}, false)
require.NoError(t, err)
require.NotEmpty(t, localIPs)
return localIPs[0]
}
func ipv6Available(t *testing.T) bool {
net, err := stdnet.NewNet()
require.NoError(t, err)
localIPs, err := localInterfaces(net, nil, nil, []NetworkType{NetworkTypeTCP6}, false)
require.NoError(t, err)
return len(localIPs) > 0
}
func TestAgentActiveTCP(t *testing.T) {
report := test.CheckRoutines(t)
defer report()
lim := test.TimeOut(time.Second * 5)
defer lim.Stop()
const listenPort = 7686
type testCase struct {
name string
networkTypes []NetworkType
listenIPAddress net.IP
selectedPairNetworkType string
}
testCases := []testCase{
{
name: "TCP4 connection",
networkTypes: []NetworkType{NetworkTypeTCP4},
listenIPAddress: getLocalIPAddress(t, NetworkTypeTCP4),
selectedPairNetworkType: tcp,
},
{
name: "UDP is preferred over TCP4", // This fails some time
networkTypes: supportedNetworkTypes(),
listenIPAddress: getLocalIPAddress(t, NetworkTypeTCP4),
selectedPairNetworkType: udp,
},
}
if ipv6Available(t) {
testCases = append(testCases,
testCase{
name: "TCP6 connection",
networkTypes: []NetworkType{NetworkTypeTCP6},
listenIPAddress: getLocalIPAddress(t, NetworkTypeTCP6),
selectedPairNetworkType: tcp,
},
testCase{
name: "UDP is preferred over TCP6", // This fails some time
networkTypes: supportedNetworkTypes(),
listenIPAddress: getLocalIPAddress(t, NetworkTypeTCP6),
selectedPairNetworkType: udp,
},
)
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
r := require.New(t)
listener, err := net.ListenTCP("tcp", &net.TCPAddr{
IP: testCase.listenIPAddress,
Port: listenPort,
})
r.NoError(err)
defer func() {
_ = listener.Close()
}()
loggerFactory := logging.NewDefaultLoggerFactory()
loggerFactory.DefaultLogLevel.Set(logging.LogLevelTrace)
tcpMux := NewTCPMuxDefault(TCPMuxParams{
Listener: listener,
Logger: loggerFactory.NewLogger("passive-ice-tcp-mux"),
ReadBufferSize: 20,
})
defer func() {
_ = tcpMux.Close()
}()
r.NotNil(tcpMux.LocalAddr(), "tcpMux.LocalAddr() is nil")
hostAcceptanceMinWait := 100 * time.Millisecond
passiveAgent, err := NewAgent(&AgentConfig{
TCPMux: tcpMux,
CandidateTypes: []CandidateType{CandidateTypeHost},
NetworkTypes: testCase.networkTypes,
LoggerFactory: loggerFactory,
IncludeLoopback: true,
HostAcceptanceMinWait: &hostAcceptanceMinWait,
})
r.NoError(err)
r.NotNil(passiveAgent)
activeAgent, err := NewAgent(&AgentConfig{
CandidateTypes: []CandidateType{CandidateTypeHost},
NetworkTypes: testCase.networkTypes,
LoggerFactory: loggerFactory,
HostAcceptanceMinWait: &hostAcceptanceMinWait,
})
r.NoError(err)
r.NotNil(activeAgent)
passiveAgentConn, activeAgenConn := connect(passiveAgent, activeAgent)
r.NotNil(passiveAgentConn)
r.NotNil(activeAgenConn)
pair := passiveAgent.getSelectedPair()
r.NotNil(pair)
r.Equal(testCase.selectedPairNetworkType, pair.Local.NetworkType().NetworkShort())
foo := []byte("foo")
_, err = passiveAgentConn.Write(foo)
r.NoError(err)
buffer := make([]byte, 1024)
n, err := activeAgenConn.Read(buffer)
r.NoError(err)
r.Equal(foo, buffer[:n])
bar := []byte("bar")
_, err = activeAgenConn.Write(bar)
r.NoError(err)
n, err = passiveAgentConn.Read(buffer)
r.NoError(err)
r.Equal(bar, buffer[:n])
r.NoError(activeAgenConn.Close())
r.NoError(passiveAgentConn.Close())
})
}
}

19
agent_config.go
View File

@@ -41,18 +41,11 @@ const (
// defaultMaxBindingRequests is the maximum number of binding requests before considering a pair failed
defaultMaxBindingRequests = 7
// TCPPriorityOffset is a number which is subtracted from the default (UDP) candidate type preference
// for host, srflx and prfx candidate types.
defaultTCPPriorityOffset = 27
// maxBufferSize is the number of bytes that can be buffered before we start to error
maxBufferSize = 1000 * 1000 // 1MB
// maxBindingRequestTimeout is the wait time before binding requests can be deleted
maxBindingRequestTimeout = 4000 * time.Millisecond
// tcpReadBufferSize is the size of the read buffer of tcpPacketConn used by active tcp candidate
tcpReadBufferSize = 8
)
func defaultCandidateTypes() []CandidateType {
@@ -181,12 +174,6 @@ type AgentConfig struct {
// Include loopback addresses in the candidate list.
IncludeLoopback bool
// TCPPriorityOffset is a number which is subtracted from the default (UDP) candidate type preference
// for host, srflx and prfx candidate types. It helps to configure relative preference of UDP candidates
// against TCP ones. Relay candidates for TCP and UDP are always 0 and not affected by this setting.
// When this is nil, defaultTCPPriorityOffset is used.
TCPPriorityOffset *uint16
}
// initWithDefaults populates an agent and falls back to defaults if fields are unset
@@ -221,12 +208,6 @@ func (config *AgentConfig) initWithDefaults(a *Agent) {
a.relayAcceptanceMinWait = *config.RelayAcceptanceMinWait
}
if config.TCPPriorityOffset == nil {
a.tcpPriorityOffset = defaultTCPPriorityOffset
} else {
a.tcpPriorityOffset = *config.TCPPriorityOffset
}
if config.DisconnectedTimeout == nil {
a.disconnectedTimeout = defaultDisconnectedTimeout
} else {

4
agent_test.go
View File

@@ -1637,7 +1637,7 @@ func TestAcceptAggressiveNomination(t *testing.T) {
KeepaliveInterval := time.Hour
cfg0 := &AgentConfig{
NetworkTypes: []NetworkType{NetworkTypeUDP4, NetworkTypeUDP6},
NetworkTypes: supportedNetworkTypes(),
MulticastDNSMode: MulticastDNSModeDisabled,
Net: net0,
@@ -1652,7 +1652,7 @@ func TestAcceptAggressiveNomination(t *testing.T) {
require.NoError(t, aAgent.OnConnectionStateChange(aNotifier))
cfg1 := &AgentConfig{
NetworkTypes: []NetworkType{NetworkTypeUDP4, NetworkTypeUDP6},
NetworkTypes: supportedNetworkTypes(),
MulticastDNSMode: MulticastDNSModeDisabled,
Net: net1,
KeepaliveInterval: &KeepaliveInterval,

61
candidate_base.go
View File

@@ -188,44 +188,6 @@ func (c *candidateBase) LocalPreference() uint16 {
return defaultLocalPreference
}
// TypePreference returns the type preference for this candidate
func (c *candidateBase) TypePreference() uint16 {
pref := c.Type().Preference()
if pref == 0 {
return 0
}
if c.NetworkType().IsTCP() {
var tcpPriorityOffset uint16 = defaultTCPPriorityOffset
if c.agent() != nil {
tcpPriorityOffset = c.agent().tcpPriorityOffset
}
pref -= tcpPriorityOffset
}
return pref
}
// Priority computes the priority for this ICE Candidate
// See: https://www.rfc-editor.org/rfc/rfc8445#section-5.1.2.1
func (c *candidateBase) Priority() uint32 {
if c.priorityOverride != 0 {
return c.priorityOverride
}
// The local preference MUST be an integer from 0 (lowest preference) to
// 65535 (highest preference) inclusive. When there is only a single IP
// address, this value SHOULD be set to 65535. If there are multiple
// candidates for a particular component for a particular data stream
// that have the same type, the local preference MUST be unique for each
// one.
return (1<<24)*uint32(c.TypePreference()) +
(1<<8)*uint32(c.LocalPreference()) +
(1<<0)*uint32(256-c.Component())
}
// RelatedAddress returns *CandidateRelatedAddress
func (c *candidateBase) RelatedAddress() *CandidateRelatedAddress {
return c.relatedAddress
@@ -301,7 +263,7 @@ func (c *candidateBase) handleInboundPacket(buf []byte, srcAddr net.Addr) {
copy(m.Raw, buf)
if err := m.Decode(); err != nil {
a.log.Warnf("Failed to handle decode ICE from %s to %s: %v", srcAddr, c.addr(), err)
a.log.Warnf("Failed to handle decode ICE from %s to %s: %v", c.addr(), srcAddr, err)
return
}
@@ -317,7 +279,7 @@ func (c *candidateBase) handleInboundPacket(buf []byte, srcAddr net.Addr) {
if !c.validateSTUNTrafficCache(srcAddr) {
remoteCandidate, valid := a.validateNonSTUNTraffic(c, srcAddr) //nolint:contextcheck
if !valid {
a.log.Warnf("Discarded message to %s, not a valid remote candidate", c.addr())
a.log.Warnf("Discarded message from %s, not a valid remote candidate", c.addr())
return
}
c.addRemoteCandidateCache(remoteCandidate, srcAddr)
@@ -381,6 +343,23 @@ func (c *candidateBase) writeTo(raw []byte, dst Candidate) (int, error) {
return n, nil
}
// Priority computes the priority for this ICE Candidate
func (c *candidateBase) Priority() uint32 {
if c.priorityOverride != 0 {
return c.priorityOverride
}
// The local preference MUST be an integer from 0 (lowest preference) to
// 65535 (highest preference) inclusive. When there is only a single IP
// address, this value SHOULD be set to 65535. If there are multiple
// candidates for a particular component for a particular data stream
// that have the same type, the local preference MUST be unique for each
// one.
return (1<<24)*uint32(c.Type().Preference()) +
(1<<8)*uint32(c.LocalPreference()) +
uint32(256-c.Component())
}
// Equal is used to compare two candidateBases
func (c *candidateBase) Equal(other Candidate) bool {
return c.NetworkType() == other.NetworkType() &&
@@ -554,7 +533,7 @@ func UnmarshalCandidate(raw string) (Candidate, error) {
case "srflx":
return NewCandidateServerReflexive(&CandidateServerReflexiveConfig{"", protocol, address, port, component, priority, foundation, relatedAddress, relatedPort})
case "prflx":
return NewCandidatePeerReflexive(&CandidatePeerReflexiveConfig{"", protocol, address, port, component, priority, foundation, relatedAddress, relatedPort, tcpType})
return NewCandidatePeerReflexive(&CandidatePeerReflexiveConfig{"", protocol, address, port, component, priority, foundation, relatedAddress, relatedPort})
case "relay":
return NewCandidateRelay(&CandidateRelayConfig{"", protocol, address, port, component, priority, foundation, relatedAddress, relatedPort, "", nil})
default:

2
candidate_peer_reflexive.go
View File

@@ -24,7 +24,6 @@ type CandidatePeerReflexiveConfig struct {
Foundation string
RelAddr string
RelPort int
TCPType TCPType
}
// NewCandidatePeerReflexive creates a new peer reflective candidate
@@ -50,7 +49,6 @@ func NewCandidatePeerReflexive(config *CandidatePeerReflexiveConfig) (*Candidate
id: candidateID,
networkType: networkType,
candidateType: CandidateTypePeerReflexive,
tcpType: config.TCPType,
address: config.Address,
port: config.Port,
resolvedAddr: createAddr(networkType, ip, config.Port),

66
candidate_test.go
View File

@@ -13,58 +13,6 @@ import (
"github.com/stretchr/testify/require"
)
func TestCandidateTypePreference(t *testing.T) {
r := require.New(t)
hostDefaultPreference := uint16(126)
prflxDefaultPreference := uint16(110)
srflxDefaultPreference := uint16(100)
relayDefaultPreference := uint16(0)
tcpOffsets := []uint16{0, 10}
for _, tcpOffset := range tcpOffsets {
agent := &Agent{
tcpPriorityOffset: tcpOffset,
}
for _, networkType := range supportedNetworkTypes() {
hostCandidate := candidateBase{
candidateType: CandidateTypeHost,
networkType: networkType,
currAgent: agent,
}
prflxCandidate := candidateBase{
candidateType: CandidateTypePeerReflexive,
networkType: networkType,
currAgent: agent,
}
srflxCandidate := candidateBase{
candidateType: CandidateTypeServerReflexive,
networkType: networkType,
currAgent: agent,
}
relayCandidate := candidateBase{
candidateType: CandidateTypeRelay,
networkType: networkType,
currAgent: agent,
}
if networkType.IsTCP() {
r.Equal(hostDefaultPreference-tcpOffset, hostCandidate.TypePreference())
r.Equal(prflxDefaultPreference-tcpOffset, prflxCandidate.TypePreference())
r.Equal(srflxDefaultPreference-tcpOffset, srflxCandidate.TypePreference())
} else {
r.Equal(hostDefaultPreference, hostCandidate.TypePreference())
r.Equal(prflxDefaultPreference, prflxCandidate.TypePreference())
r.Equal(srflxDefaultPreference, srflxCandidate.TypePreference())
}
r.Equal(relayDefaultPreference, relayCandidate.TypePreference())
}
}
}
func TestCandidatePriority(t *testing.T) {
for _, test := range []struct {
Candidate Candidate
@@ -88,7 +36,7 @@ func TestCandidatePriority(t *testing.T) {
tcpType: TCPTypeActive,
},
},
WantPriority: 1675624447,
WantPriority: 2128609279,
},
{
Candidate: &CandidateHost{
@@ -99,7 +47,7 @@ func TestCandidatePriority(t *testing.T) {
tcpType: TCPTypePassive,
},
},
WantPriority: 1671430143,
WantPriority: 2124414975,
},
{
Candidate: &CandidateHost{
@@ -110,7 +58,7 @@ func TestCandidatePriority(t *testing.T) {
tcpType: TCPTypeSimultaneousOpen,
},
},
WantPriority: 1667235839,
WantPriority: 2120220671,
},
{
Candidate: &CandidatePeerReflexive{
@@ -130,7 +78,7 @@ func TestCandidatePriority(t *testing.T) {
tcpType: TCPTypeSimultaneousOpen,
},
},
WantPriority: 1407188991,
WantPriority: 1860173823,
},
{
Candidate: &CandidatePeerReflexive{
@@ -141,7 +89,7 @@ func TestCandidatePriority(t *testing.T) {
tcpType: TCPTypeActive,
},
},
WantPriority: 1402994687,
WantPriority: 1855979519,
},
{
Candidate: &CandidatePeerReflexive{
@@ -152,7 +100,7 @@ func TestCandidatePriority(t *testing.T) {
tcpType: TCPTypePassive,
},
},
WantPriority: 1398800383,
WantPriority: 1851785215,
},
{
Candidate: &CandidateServerReflexive{
@@ -337,7 +285,7 @@ func TestCandidateMarshal(t *testing.T) {
},
"",
},
"1052353102 1 tcp 1675624447 192.168.0.196 0 typ host tcptype active",
"1052353102 1 tcp 2128609279 192.168.0.196 0 typ host tcptype active",
false,
},
{

3
candidatetype.go
View File

@@ -48,9 +48,8 @@ func (c CandidateType) Preference() uint16 {
return 100
case CandidateTypeRelay, CandidateTypeUnspecified:
return 0
default:
return 0
}
return 0
}
func containsCandidateType(candidateType CandidateType, candidateTypeList []CandidateType) bool {

95
gather.go
View File

@@ -25,12 +25,6 @@ const (
stunGatherTimeout = time.Second * 5
)
type connConfig struct {
conn net.PacketConn
port int
tcpType TCPType
}
// Close a net.Conn and log if we have a failure
func closeConnAndLog(c io.Closer, log logging.LeveledLogger, msg string, args ...interface{}) {
if c == nil || (reflect.ValueOf(c).Kind() == reflect.Ptr && reflect.ValueOf(c).IsNil()) {
@@ -161,21 +155,53 @@ func (a *Agent) gatherCandidatesLocal(ctx context.Context, networkTypes []Networ
}
for network := range networks {
var connConfigs []connConfig
type connAndPort struct {
conn net.PacketConn
port int
}
var (
conns []connAndPort
tcpType TCPType
)
switch network {
case tcp:
// Handle ICE TCP active mode
connConfigs = append(connConfigs, connConfig{nil, 0, TCPTypeActive})
if a.tcpMux == nil {
continue
}
// Handle ICE TCP passive mode
if a.tcpMux != nil {
connConfigs = a.getTCPMuxConns(mappedIP, ip, network, connConfigs)
var muxConns []net.PacketConn
if multi, ok := a.tcpMux.(AllConnsGetter); ok {
a.log.Debugf("GetAllConns by ufrag: %s", a.localUfrag)
muxConns, err = multi.GetAllConns(a.localUfrag, mappedIP.To4() == nil, ip)
if err != nil {
a.log.Warnf("Failed to get all TCP connections by ufrag: %s %s %s", network, ip, a.localUfrag)
continue
}
} else {
a.log.Debugf("GetConn by ufrag: %s", a.localUfrag)
conn, err := a.tcpMux.GetConnByUfrag(a.localUfrag, mappedIP.To4() == nil, ip)
if err != nil {
a.log.Warnf("Failed to get TCP connections by ufrag: %s %s %s", network, ip, a.localUfrag)
continue
}
muxConns = []net.PacketConn{conn}
}
if len(connConfigs) == 0 {
// Extract the port for each PacketConn we got.
for _, conn := range muxConns {
if tcpConn, ok := conn.LocalAddr().(*net.TCPAddr); ok {
conns = append(conns, connAndPort{conn, tcpConn.Port})
} else {
a.log.Warnf("Failed to get port of connection from TCPMux: %s %s %s", network, ip, a.localUfrag)
}
}
if len(conns) == 0 {
// Didn't succeed with any, try the next network.
continue
}
tcpType = TCPTypePassive
// Is there a way to verify that the listen address is even
// accessible from the current interface.
case udp:
@@ -186,36 +212,36 @@ func (a *Agent) gatherCandidatesLocal(ctx context.Context, networkTypes []Networ
}
if udpConn, ok := conn.LocalAddr().(*net.UDPAddr); ok {
connConfigs = append(connConfigs, connConfig{conn, udpConn.Port, TCPTypeUnspecified})
conns = append(conns, connAndPort{conn, udpConn.Port})
} else {
a.log.Warnf("Failed to get port of UDPAddr from ListenUDPInPortRange: %s %s %s", network, ip, a.localUfrag)
continue
}
}
for _, connConfig := range connConfigs {
for _, connAndPort := range conns {
hostConfig := CandidateHostConfig{
Network: network,
Address: address,
Port: connConfig.port,
Port: connAndPort.port,
Component: ComponentRTP,
TCPType: connConfig.tcpType,
TCPType: tcpType,
}
c, err := NewCandidateHost(&hostConfig)
if err != nil {
closeConnAndLog(connConfig.conn, a.log, "failed to create host candidate: %s %s %d: %v", network, mappedIP, connConfig.port, err)
closeConnAndLog(connAndPort.conn, a.log, "failed to create host candidate: %s %s %d: %v", network, mappedIP, connAndPort.port, err)
continue
}
if a.mDNSMode == MulticastDNSModeQueryAndGather {
if err = c.setIP(ip); err != nil {
closeConnAndLog(connConfig.conn, a.log, "failed to create host candidate: %s %s %d: %v", network, mappedIP, connConfig.port, err)
closeConnAndLog(connAndPort.conn, a.log, "failed to create host candidate: %s %s %d: %v", network, mappedIP, connAndPort.port, err)
continue
}
}
if err := a.addCandidate(ctx, c, connConfig.conn); err != nil {
if err := a.addCandidate(ctx, c, connAndPort.conn); err != nil {
if closeErr := c.close(); closeErr != nil {
a.log.Warnf("Failed to close candidate: %v", closeErr)
}
@@ -226,37 +252,6 @@ func (a *Agent) gatherCandidatesLocal(ctx context.Context, networkTypes []Networ
}
}
func (a *Agent) getTCPMuxConns(mappedIP net.IP, ip net.IP, network string, conns []connConfig) []connConfig {
var muxConns []net.PacketConn
if multi, ok := a.tcpMux.(AllConnsGetter); ok {
a.log.Debugf("GetAllConns by ufrag: %s", a.localUfrag)
var err error
muxConns, err = multi.GetAllConns(a.localUfrag, mappedIP.To4() == nil, ip)
if err != nil {
a.log.Warnf("Failed to get all TCP connections by ufrag: %s %s %s", network, ip, a.localUfrag)
return conns
}
} else {
a.log.Debugf("GetConn by ufrag: %s", a.localUfrag)
conn, err := a.tcpMux.GetConnByUfrag(a.localUfrag, mappedIP.To4() == nil, ip)
if err != nil {
a.log.Warnf("Failed to get TCP connections by ufrag: %s %s %s", network, ip, a.localUfrag)
return conns
}
muxConns = []net.PacketConn{conn}
}
// Extract the port for each PacketConn we got.
for _, conn := range muxConns {
if tcpConn, ok := conn.LocalAddr().(*net.TCPAddr); ok {
conns = append(conns, connConfig{conn, tcpConn.Port, TCPTypePassive})
} else {
a.log.Warnf("Failed to get port of connection from TCPMux: %s %s %s", network, ip, a.localUfrag)
}
}
return conns
}
func (a *Agent) gatherCandidatesLocalUDPMux(ctx context.Context) error { //nolint:gocognit
if a.udpMux == nil {
return errUDPMuxDisabled

5
gather_test.go
View File

@@ -675,7 +675,7 @@ func TestMultiUDPMuxUsage(t *testing.T) {
}
a, err := NewAgent(&AgentConfig{
NetworkTypes: []NetworkType{NetworkTypeUDP4, NetworkTypeUDP6},
NetworkTypes: supportedNetworkTypes(),
CandidateTypes: []CandidateType{CandidateTypeHost},
UDPMux: NewMultiUDPMuxDefault(udpMuxInstances...),
})
@@ -751,8 +751,7 @@ func TestMultiTCPMuxUsage(t *testing.T) {
portFound := make(map[int]bool)
for c := range candidateCh {
activeCandidate := c.Port() == 0
if c.NetworkType().IsTCP() && !activeCandidate {
if c.NetworkType().IsTCP() {
portFound[c.Port()] = true
}
}