Remove global state for ICE TCP

This addresses a few points issue of #245:

 - Take a net.Listener instead of having global state
 - Expose a net.TCPMux based API

Also, the unused closeChannel was removed from tcp_mux.go

Closes #253.

agent.go

@@ -120,8 +120,8 @@ type Agent struct {
 	loggerFactory logging.LoggerFactory
 	log           logging.LeveledLogger
 
-	net *vnet.Net
-	tcp *tcpIPMux
+	net    *vnet.Net
+	tcpMux TCPMux
 
 	interfaceFilter func(string) bool
 
@@ -306,11 +306,10 @@ func NewAgent(config *AgentConfig) (*Agent, error) {
 		insecureSkipVerify: config.InsecureSkipVerify,
 	}
 
-	a.tcp = newTCPIPMux(tcpIPMuxParams{
-		ListenPort:     config.TCPListenPort,
-		Logger:         log,
-		ReadBufferSize: 8,
-	})
+	a.tcpMux = config.TCPMux
+	if a.tcpMux == nil {
+		a.tcpMux = newInvalidTCPMux()
+	}
 
 	if a.net == nil {
 		a.net = vnet.NewNet(nil)
@@ -887,7 +886,9 @@ func (a *Agent) Close() error {
 
 	a.gatherCandidateCancel()
 	a.err.Store(ErrClosed)
-	a.tcp.RemoveUfrag(a.localUfrag)
+
+	a.tcpMux.RemoveConnByUfrag(a.localUfrag)
+
 	close(a.done)
 
 	<-done

agent_config.go

@@ -139,10 +139,10 @@ type AgentConfig struct {
 	// to TURN servers via TLS or DTLS
 	InsecureSkipVerify bool
 
-	// TCPListenPort will be used to start a TCP listener on all allowed interfaces for
-	// ICE TCP. Currently only passive candidates are supported. This functionality is
-	// experimental and this API will likely change in the future.
-	TCPListenPort int
+	// TCPMux will be used for multiplexing incoming TCP connections for ICE TCP.
+	// Currently only passive candidates are supported. This functionality is
+	// experimental and the API might change in the future.
+	TCPMux TCPMux
 }
 
 // initWithDefaults populates an agent and falls back to defaults if fields are unset

errors.go

@@ -103,6 +103,8 @@ var (
 	// ErrRunCanceled indicates a run operation was canceled by its individual done
 	ErrRunCanceled = errors.New("run was canceled by done")
 
+	ErrTCPMuxNotInitialized = errors.New("TCPMux is not initialized")
+
 	// ErrTCPRemoteAddrAlreadyExists indicates we already have the connection with same remote addr.
 	ErrTCPRemoteAddrAlreadyExists = errors.New("conn with same remote addr already exists")
 )

gather.go

@@ -161,28 +161,22 @@ func (a *Agent) gatherCandidatesLocal(ctx context.Context, networkTypes []Networ
 			var tcpType TCPType
 			switch network {
 			case tcp:
-				if a.tcp == nil {
-					continue
-				}
-
-				// below is for passive mode
+				// Handle ICE TCP passive mode
 				// TODO active mode
 				// TODO S-O mode
 
-				mux, muxErr := a.tcp.Listen(ip)
-				if muxErr != nil {
-					a.log.Warnf("could not listen %s %s\n", network, ip)
-					continue
-				}
-
 				a.log.Debugf("GetConn by ufrag: %s\n", a.localUfrag)
-				conn, err = mux.GetConn(a.localUfrag)
+				conn, err = a.tcpMux.GetConnByUfrag(a.localUfrag)
 				if err != nil {
-					a.log.Warnf("error getting tcp conn by ufrag: %s %s\n", network, ip, a.localUfrag)
+					if err != ErrTCPMuxNotInitialized {
+						a.log.Warnf("error getting tcp conn by ufrag: %s %s\n", network, ip, a.localUfrag)
+					}
 					continue
 				}
 				port = conn.LocalAddr().(*net.TCPAddr).Port
 				tcpType = TCPTypePassive
+				// TODO is there a way to verify that the listen address is even
+				// accessible from the current interface.
 			case udp:
 				conn, err = listenUDPInPortRange(a.net, a.log, int(a.portmax), int(a.portmin), network, &net.UDPAddr{IP: ip, Port: 0})
 				if err != nil {

gather_test.go

@@ -14,9 +14,11 @@ import (
 
 	"github.com/pion/dtls/v2"
 	"github.com/pion/dtls/v2/pkg/crypto/selfsign"
+	"github.com/pion/logging"
 	"github.com/pion/transport/test"
 	"github.com/pion/turn/v2"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestListenUDP(t *testing.T) {
@@ -116,11 +118,25 @@ func TestSTUNConcurrency(t *testing.T) {
 		Port:   serverPort,
 	})
 
+	listener, err := net.ListenTCP("tcp", &net.TCPAddr{
+		IP: net.IP{127, 0, 0, 1},
+	})
+	require.NoError(t, err)
+	defer func() {
+		_ = listener.Close()
+	}()
+
 	a, err := NewAgent(&AgentConfig{
 		NetworkTypes:   supportedNetworkTypes,
 		Urls:           urls,
 		CandidateTypes: []CandidateType{CandidateTypeHost, CandidateTypeServerReflexive},
-		TCPListenPort:  9999,
+		TCPMux: NewTCPMuxDefault(
+			TCPMuxParams{
+				Listener:       listener,
+				Logger:         logging.NewDefaultLoggerFactory().NewLogger("ice"),
+				ReadBufferSize: 8,
+			},
+		),
 	})
 	assert.NoError(t, err)