TURN client refactor

Implemented allocation & permission refresh
Implemented permissions
Implemented channel binding
Threa-safety improvements
Optimized inbound packet demuxing
Resolves #74

client_test.go

@@ -3,40 +3,101 @@ package turn
 import (
 	"net"
 	"testing"
+	"time"
 
 	"github.com/pion/logging"
+	"github.com/stretchr/testify/assert"
 )
 
-func TestClient(t *testing.T) {
+func createListeningTestClient(t *testing.T, loggerFactory logging.LoggerFactory) (*Client, bool) {
+	conn, err := net.ListenPacket("udp4", "0.0.0.0:0")
+	if !assert.NoError(t, err, "should succeed") {
+		return nil, false
+	}
+	c, err := NewClient(&ClientConfig{
+		Conn:          conn,
+		LoggerFactory: loggerFactory,
+	})
+	if !assert.NoError(t, err, "should succeed") {
+		return nil, false
+	}
+	err = c.Listen()
+	if !assert.NoError(t, err, "should succeed") {
+		return nil, false
+	}
+
+	return c, true
+}
+
+func createListeningTestClientWithSTUNServ(t *testing.T, loggerFactory logging.LoggerFactory) (*Client, bool) {
+	conn, err := net.ListenPacket("udp4", "0.0.0.0:0")
+	if !assert.NoError(t, err, "should succeed") {
+		return nil, false
+	}
+	c, err := NewClient(&ClientConfig{
+		STUNServerAddr: "stun1.l.google.com:19302",
+		Conn:           conn,
+
+		LoggerFactory: loggerFactory,
+	})
+	if !assert.NoError(t, err, "should succeed") {
+		return nil, false
+	}
+	err = c.Listen()
+	if !assert.NoError(t, err, "should succeed") {
+		return nil, false
+	}
+
+	return c, true
+}
+
+func TestClientWithSTUN(t *testing.T) {
 	loggerFactory := logging.NewDefaultLoggerFactory()
+	log := loggerFactory.NewLogger("test")
 
-	t.Run("SendSTUNRequest Parallel", func(t *testing.T) {
-		c, err := NewClient(&ClientConfig{
-			ListeningAddress: "0.0.0.0:0",
-			LoggerFactory:    loggerFactory,
-		})
-		if err != nil {
-			t.Fatal(err)
+	t.Run("SendBindingRequest", func(t *testing.T) {
+		c, ok := createListeningTestClientWithSTUNServ(t, loggerFactory)
+		if !ok {
+			return
 		}
+		defer c.Close()
+
+		resp, err := c.SendBindingRequest()
+		assert.NoError(t, err, "should succeed")
+		log.Debugf("mapped-addr: %s", resp.String())
+		assert.Equal(t, 0, c.trMap.Size(), "should be no transaction left")
+	})
+
+	t.Run("SendBindingRequestTo Parallel", func(t *testing.T) {
+		c, ok := createListeningTestClient(t, loggerFactory)
+		if !ok {
+			return
+		}
+		defer c.Close()
 
 		// simple channel fo go routine start signaling
 		started := make(chan struct{})
 		finished := make(chan struct{})
 		var err1 error
 		var resp1 interface{}
 
+		to, err := net.ResolveUDPAddr("udp4", "stun1.l.google.com:19302")
+		if !assert.NoError(t, err, "should succeed") {
+			return
+		}
+
 		// stun1.l.google.com:19302, more at https://gist.github.com/zziuni/3741933#file-stuns-L5
 		go func() {
 			close(started)
-			resp1, err1 = c.SendSTUNRequest(net.IPv4(74, 125, 143, 127), 19302)
+			resp1, err1 = c.SendBindingRequestTo(to)
 			close(finished)
 		}()
 
 		// block until go routine is started to make two almost parallel requests
 
 		<-started
 
-		resp2, err2 := c.SendSTUNRequest(net.IPv4(74, 125, 143, 127), 19302)
+		resp2, err2 := c.SendBindingRequestTo(to)
 		if err2 != nil {
 			t.Fatal(err)
 		} else {
@@ -51,27 +112,28 @@ func TestClient(t *testing.T) {
 		}
 	})
 
-	t.Run("Listen error", func(t *testing.T) {
+	t.Run("NewClient should fail if Conn is nil", func(t *testing.T) {
 		_, err := NewClient(&ClientConfig{
-			ListeningAddress: "255.255.255.256:65535",
-			LoggerFactory:    loggerFactory,
+			LoggerFactory: loggerFactory,
 		})
-		if err == nil {
-			t.Fatal("listening on 255.255.255.256:65535 should fail")
-		}
+		assert.Error(t, err, "should fail")
 	})
 
-	/*
-		// Unable to perform this test atm because there is no timeout and the test may run infinitely
-		t.Run("SendSTUNRequest timeout", func(t *testing.T) {
-			c, err := NewClient("0.0.0.0:0")
-			if err != nil {
-				t.Fatal(err)
-			}
-			_, err = c.SendSTUNRequest(net.IPv4(255, 255, 255, 255), 65535)
-			if err == nil {
-				t.Fatal("request to 255.255.255.255:65535 should fail")
-			}
-		})
-	*/
+	t.Run("SendBindingRequestTo timeout", func(t *testing.T) {
+		c, ok := createListeningTestClient(t, loggerFactory)
+		if !ok {
+			return
+		}
+		defer c.Close()
+
+		to, err := net.ResolveUDPAddr("udp4", "127.0.0.1:9")
+		if !assert.NoError(t, err, "should succeed") {
+			return
+		}
+
+		c.rto = 10 * time.Millisecond // force short timeout
+
+		_, err = c.SendBindingRequestTo(to)
+		log.Debug(err.Error())
+	})
 }

cmd/client/main.go

@@ -1,8 +1,8 @@
 package main
 
 import (
-	"errors"
 	"flag"
+	"fmt"
 	"log"
 	"net"
 
@@ -11,24 +11,37 @@ import (
 )
 
 func main() {
-	host := flag.String("host", "74.125.143.127", "IP of TURN Server. Default is the IP of stun1.l.google.com.")
+	//host := flag.String("host", "74.125.143.127", "IP of TURN Server. Default is the IP of stun1.l.google.com.")
+	host := flag.String("host", "stun1.l.google.com", "IP of TURN Server. Default is stun1.l.google.com.")
 	port := flag.Int("port", 19302, "Port of TURN server.")
 	flag.Parse()
 
-	ip := net.ParseIP(*host)
-	if ip == nil {
-		panic(errors.New("failed to parse host IP"))
+	conn, err := net.ListenPacket("udp4", "0.0.0.0:0")
+	if err != nil {
+		panic(err)
 	}
+	defer func() {
+		if err2 := conn.Close(); err2 != nil {
+			panic(err2)
+		}
+	}()
 
 	c, err := turn.NewClient(&turn.ClientConfig{
-		ListeningAddress: "0.0.0.0:0",
-		LoggerFactory:    logging.NewDefaultLoggerFactory(),
+		STUNServerAddr: fmt.Sprintf("%s:%d", *host, *port),
+		Conn:           conn,
+		LoggerFactory:  logging.NewDefaultLoggerFactory(),
 	})
 	if err != nil {
 		panic(err)
 	}
+	defer c.Close()
+
+	err = c.Listen()
+	if err != nil {
+		panic(err)
+	}
 
-	mappedAddr, err := c.SendSTUNRequest(ip, *port)
+	mappedAddr, err := c.SendBindingRequest()
 	if err != nil {
 		panic(err)
 	}

go.mod

@@ -5,9 +5,9 @@ go 1.12
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/gortc/turn v0.8.0
-	github.com/pion/logging v0.2.1
+	github.com/pion/logging v0.2.2
 	github.com/pion/stun v0.3.1
-	github.com/pion/transport v0.8.1
+	github.com/pion/transport v0.8.4
 	github.com/pkg/errors v0.8.1
 	github.com/stretchr/testify v1.3.0
 	golang.org/x/net v0.0.0-20190403144856-b630fd6fe46b

go.sum

@@ -5,10 +5,12 @@ github.com/gortc/turn v0.8.0 h1:WWQi1jkoPmc2E7qgUMcZleveKikT9Ksi3QGIl8ZtY3Q=
 github.com/gortc/turn v0.8.0/go.mod h1:gvguwaGAFyv5/9KrcW9MkCgHALYD+e99mSM7pSCYYho=
 github.com/pion/logging v0.2.1 h1:LwASkBKZ+2ysGJ+jLv1E/9H1ge0k1nTfi1X+5zirkDk=
 github.com/pion/logging v0.2.1/go.mod h1:k0/tDVsRCX2Mb2ZEmTqNa7CWsQPc+YYCB7Q+5pahoms=
+github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY=
+github.com/pion/logging v0.2.2/go.mod h1:k0/tDVsRCX2Mb2ZEmTqNa7CWsQPc+YYCB7Q+5pahoms=
 github.com/pion/stun v0.3.1 h1:d09JJzOmOS8ZzIp8NppCMgrxGZpJ4Ix8qirfNYyI3BA=
 github.com/pion/stun v0.3.1/go.mod h1:xrCld6XM+6GWDZdvjPlLMsTU21rNxnO6UO8XsAvHr/M=
-github.com/pion/transport v0.8.1 h1:FUHJFd4MaIEJmlpiGx+ZH8j9JLsERnROHQPA9zNFFAs=
-github.com/pion/transport v0.8.1/go.mod h1:nAmRRnn+ArVtsoNuwktvAD+jrjSD7pA+H3iRmZwdUno=
+github.com/pion/transport v0.8.4 h1:Wios3j8IFmrli4pHiXhGMVnj1DYWiukcboZGSv8kj2M=
+github.com/pion/transport v0.8.4/go.mod h1:nAmRRnn+ArVtsoNuwktvAD+jrjSD7pA+H3iRmZwdUno=
 github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=

internal/allocation/allocation.go

@@ -253,7 +253,9 @@ func (a *Allocation) packetHandler(m *Manager) {
 			if err != nil {
 				a.log.Errorf("Failed to send DataIndication from allocation %v %v", srcAddr, err)
 			}
-			a.log.Debugf("relaying message to client at %s", a.fiveTuple.SrcAddr.String())
+			a.log.Debugf("relaying message from %s to client at %s",
+				srcAddr.String(),
+				a.fiveTuple.SrcAddr.String())
 			if _, err = a.TurnSocket.WriteTo(msg.Raw, a.fiveTuple.SrcAddr); err != nil {
 				a.log.Errorf("Failed to send DataIndication from allocation %v %v", srcAddr, err)
 			}

internal/client/atomic_bool.go

@@ -0,0 +1,39 @@
+package client
+
+import (
+	"sync/atomic"
+)
+
+// AtomicBool is an atomic boolean struct
+type AtomicBool struct {
+	n int32
+}
+
+// NewAtomicBool creates a new instance of AtomicBool
+func NewAtomicBool(initiallyTrue bool) *AtomicBool {
+	var n int32
+	if initiallyTrue {
+		n = 1
+	}
+	return &AtomicBool{n: n}
+}
+
+// SetToTrue sets this value to true
+func (b *AtomicBool) SetToTrue() {
+	atomic.StoreInt32(&b.n, 1)
+}
+
+// SetToFalse sets this value to false
+func (b *AtomicBool) SetToFalse() {
+	atomic.StoreInt32(&b.n, 0)
+}
+
+// True returns true if it is set to true
+func (b *AtomicBool) True() bool {
+	return atomic.LoadInt32(&b.n) != int32(0)
+}
+
+// False return true if it is set to false
+func (b *AtomicBool) False() bool {
+	return atomic.LoadInt32(&b.n) == int32(0)
+}

internal/client/atomic_bool_test.go

@@ -0,0 +1,24 @@
+package client
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestAtomicBool(t *testing.T) {
+	b0 := NewAtomicBool(false)
+	assert.False(t, b0.True(), "should false")
+	assert.True(t, b0.False(), "should false")
+
+	b1 := NewAtomicBool(true)
+	assert.True(t, b1.True(), "should true")
+	assert.False(t, b1.False(), "should true")
+
+	b0.SetToTrue()
+	assert.True(t, b0.True(), "should true")
+	assert.False(t, b0.False(), "should true")
+	b0.SetToFalse()
+	assert.False(t, b0.True(), "should false")
+	assert.True(t, b0.False(), "should false")
+}