Repository: pion/webrtc (go)

Issue:
Enable bufferedAmount and bufferedAmountLowThreshold
## Summary
The work at pions/sctp#11 enables congestion control and "bufferredAmount" is now available in the work. 

## Motivation
The following webrtc API methods need to be writed to SCTP layer.
* DataChannel.BufferredAmount()
* DataChannel.BufferedAmountLowThreshold()
* DataChannel.SetBufferedAmountLowThreshold()
* DataChannel.OnBufferedAmountLow()

## Additional context
See discussions in pions/sctp#11 and pions/sctp#23


Pull request description:
[WIP] Impleted bufferedAmount interface
#### Description
* T3-rtx&congestion control is implmented at sctp@t3rtx-timer
* Enabled the following methods
   - DataChannel.BufferredAmount()
   - DataChannel.BufferedAmountLowThreshold()
   - DataChannel.SetBufferedAmountLowThreshold()
   - DataChannel.OnBufferedAmountLow()
* Add tests for these methods

#### Reference issue
Fixes #569


Patch:
diff --git a/datachannel.go b/datachannel.go
index 8f69109f..276c10fa 100644
--- a/datachannel.go
+++ b/datachannel.go
@@ -39,12 +39,12 @@ type DataChannel struct {
 	// "blob". This attribute controls how binary data is exposed to scripts.
 	// binaryType                 string
 
-	// OnBufferedAmountLow func()
 	// OnError             func()
 
-	onMessageHandler func(DataChannelMessage)
-	onOpenHandler    func()
-	onCloseHandler   func()
+	onMessageHandler    func(DataChannelMessage)
+	onOpenHandler       func()
+	onCloseHandler      func()
+	onBufferedAmountLow func()
 
 	sctpTransport *SCTPTransport
 	dataChannel   *datachannel.DataChannel
@@ -133,6 +133,7 @@ func (d *DataChannel) open(sctpTransport *SCTPTransport) error {
 		Priority:             datachannel.ChannelPriorityNormal, // TODO: Wiring
 		ReliabilityParameter: reliabilityParameteer,
 		Label:                d.label,
+		LoggerFactory:        d.api.settingEngine.LoggerFactory,
 	}
 
 	dc, err := datachannel.Dial(d.sctpTransport.association, *d.id, cfg)
@@ -141,6 +142,10 @@ func (d *DataChannel) open(sctpTransport *SCTPTransport) error {
 		return err
 	}
 
+	// bufferedAmountLowThreshold and onBufferedAmountLow might be set earlier
+	dc.SetBufferedAmountLowThreshold(d.bufferedAmountLowThreshold)
+	dc.OnBufferedAmountLow(d.onBufferedAmountLow)
+
 	d.readyState = DataChannelStateOpen
 	d.mu.Unlock()
 
@@ -458,8 +463,10 @@ func (d *DataChannel) BufferedAmount() uint64 {
 	d.mu.RLock()
 	defer d.mu.RUnlock()
 
-	// TODO: wire to SCTP (pion/sctp#11)
-	return 0
+	if d.dataChannel == nil {
+		return 0
+	}
+	return d.dataChannel.BufferedAmount()
 }
 
 // BufferedAmountLowThreshold represents the threshold at which the
@@ -467,23 +474,39 @@ func (d *DataChannel) BufferedAmount() uint64 {
 // from above this threshold to equal or below it, the bufferedamountlow
 // event fires. BufferedAmountLowThreshold is initially zero on each new
 // DataChannel, but the application may change its value at any time.
+// The threshold is set to 0 by default.
 func (d *DataChannel) BufferedAmountLowThreshold() uint64 {
 	d.mu.RLock()
 	defer d.mu.RUnlock()
 
-	// TODO: wire to SCTP (pion/sctp#11)
-	return d.bufferedAmountLowThreshold
+	if d.dataChannel == nil {
+		return d.bufferedAmountLowThreshold
+	}
+	return d.dataChannel.BufferedAmountLowThreshold()
 }
 
-// SetBufferedAmountLowThreshold represents the threshold at which the
-// bufferedAmount is considered to be low. When the bufferedAmount decreases
-// from above this threshold to equal or below it, the bufferedamountlow
-// event fires. BufferedAmountLowThreshold is initially zero on each new
-// DataChannel, but the application may change its value at any time.
+// SetBufferedAmountLowThreshold is used to update the threshold.
+// See BufferedAmountLowThreshold().
 func (d *DataChannel) SetBufferedAmountLowThreshold(th uint64) {
 	d.mu.Lock()
 	defer d.mu.Unlock()
 
-	// TODO: wire to SCTP (pion/sctp#11)
 	d.bufferedAmountLowThreshold = th
+
+	if d.dataChannel != nil {
+		d.dataChannel.SetBufferedAmountLowThreshold(th)
+	}
+}
+
+// OnBufferedAmountLow sets an event handler which is invoked when
+// the number of bytes of outgoing data becomes lower than the
+// BufferedAmountLowThreshold.
+func (d *DataChannel) OnBufferedAmountLow(f func()) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	d.onBufferedAmountLow = f
+	if d.dataChannel != nil {
+		d.dataChannel.OnBufferedAmountLow(f)
+	}
 }
diff --git a/go.mod b/go.mod
index 896eb0ca..affe8566 100644
--- a/go.mod
+++ b/go.mod
@@ -3,14 +3,14 @@ module github.com/pion/webrtc
 go 1.12
 
 require (
-	github.com/pion/datachannel v1.2.2
+	github.com/pion/datachannel v1.3.0
 	github.com/pion/dtls v1.2.3
 	github.com/pion/ice v0.2.1
 	github.com/pion/logging v0.2.1
 	github.com/pion/quic v0.1.1
 	github.com/pion/rtcp v1.1.4
 	github.com/pion/rtp v1.1.1
-	github.com/pion/sctp v1.4.4
+	github.com/pion/sctp v1.5.0
 	github.com/pion/sdp/v2 v2.1.1
 	github.com/pion/srtp v1.1.3
 	github.com/pion/transport v0.6.0
diff --git a/go.sum b/go.sum
index 54a50bc7..d54d3cfa 100644
--- a/go.sum
+++ b/go.sum
@@ -20,12 +20,13 @@ github.com/onsi/ginkgo v1.7.0 h1:WSHQ+IS43OoUrWtD1/bbclrwK8TTH5hzp+umCiuxHgs=
 github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/gomega v1.4.3 h1:RE1xgDvH7imwFD45h+u2SgIfERHlS2yNG4DObb5BSKU=
 github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
-github.com/pion/datachannel v1.2.2 h1:Ir4NL3svXjix26skf3Y0yfDXC6ztB5PnwB5EacfV/qY=
-github.com/pion/datachannel v1.2.2/go.mod h1:OHz0QB98N/iLKf38LwSA8m9A6Y51cswmp/IluvjsPWk=
+github.com/pion/datachannel v1.3.0 h1:gxt/xGufDn8Yylk0uJB231xbGQVlFjVps+KdUAUl5Ls=
+github.com/pion/datachannel v1.3.0/go.mod h1:lxFbZLIT+EBPmy5AiCv8M0CXkcuTL53A4cyagZiRrDo=
 github.com/pion/dtls v1.2.3 h1:4ZVFJHLIvro3holCH7bs1XhuAzs9idC8gUsMRDfahwM=
 github.com/pion/dtls v1.2.3/go.mod h1:krdyIQPgcrKJu/Rr/lJS9tIFeA6hqMpcpDeVLzLN0+s=
 github.com/pion/ice v0.2.1 h1:DhYn8s52H54SBbS5qu3XoGvTfseU47pe15yV3udNpww=
 github.com/pion/ice v0.2.1/go.mod h1:igvbO76UeYthbSu0UsUTqjyWpFT3diUmM+x2vt4p4fw=
+github.com/pion/logging v0.2.1-0.20190404202522-3c79a8accd0a/go.mod h1:k0/tDVsRCX2Mb2ZEmTqNa7CWsQPc+YYCB7Q+5pahoms=
 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/quic v0.1.1 h1:D951FV+TOqI9A0rTF7tHx0Loooqz+nyzjEyj8o3PuMA=
@@ -34,8 +35,8 @@ github.com/pion/rtcp v1.1.4 h1:LW1zS49QXeuhKyFr7hFOlvnZZ2cR8lluTYg5jilGnKY=
 github.com/pion/rtcp v1.1.4/go.mod h1:a5dj2d6BKIKHl43EnAOIrCczcjESrtPuMgfmL6/K6QM=
 github.com/pion/rtp v1.1.1 h1:lag+9/lSOLBEYeYB/28KXm/ka1H++4wkmSj/WkttV6Y=
 github.com/pion/rtp v1.1.1/go.mod h1:/l4cvcKd0D3u9JLs2xSVI95YkfXW87a3br3nqmVtSlE=
-github.com/pion/sctp v1.4.4 h1:0d3c1g7OtvPYMVWPSNu8w37yJaKUcjCupVQ1zVcnIOc=
-github.com/pion/sctp v1.4.4/go.mod h1:oS057xrxwNbgFWap0buUuAJalfVezrOEwdmEoydf5x4=
+github.com/pion/sctp v1.5.0 h1:VcixluIP/XBKL3wRRYIzpvbkFQFVs2yUWJo1NUivy7k=
+github.com/pion/sctp v1.5.0/go.mod h1:btfZTRxsoVwp7PfvorgOKqkxV/BKHGGrNf1YUKnMGRQ=
 github.com/pion/sdp/v2 v2.1.1 h1:i3fAyjiLuQseYNo0BtCOPfzp91Ppb7vasRGmUUTog28=
 github.com/pion/sdp/v2 v2.1.1/go.mod h1:idSlWxhfWQDtTy9J05cgxpHBu/POwXN2VDRGYxT/EjU=
 github.com/pion/srtp v1.1.3 h1:8VNvpIft38BDgQj6pMSjjQnzd3KyMbheSFnJplKmIDU=
diff --git a/sctptransport.go b/sctptransport.go
index ac25d20c..934d2b63 100644
--- a/sctptransport.go
+++ b/sctptransport.go
@@ -89,7 +89,10 @@ func (r *SCTPTransport) Start(remoteCaps SCTPCapabilities) error {
 		return err
 	}
 
-	sctpAssociation, err := sctp.Client(r.dtlsTransport.conn)
+	sctpAssociation, err := sctp.Client(sctp.Config{
+		NetConn:       r.dtlsTransport.conn,
+		LoggerFactory: r.api.settingEngine.LoggerFactory,
+	})
 	if err != nil {
 		return err
 	}
@@ -132,7 +135,9 @@ func (r *SCTPTransport) acceptDataChannels() {
 	a := r.association
 	r.lock.RUnlock()
 	for {
-		dc, err := datachannel.Accept(a)
+		dc, err := datachannel.Accept(a, &datachannel.Config{
+			LoggerFactory: r.api.settingEngine.LoggerFactory,
+		})
 		if err != nil {
 			r.log.Errorf("Failed to accept data channel: %v", err)
 			// TODO: Kill DataChannel/PeerConnection?


Test patch:
diff --git a/datachannel_go_test.go b/datachannel_go_test.go
index cef3d4ca..43d4ef33 100644
--- a/datachannel_go_test.go
+++ b/datachannel_go_test.go
@@ -195,3 +195,153 @@ func TestDataChannelParamters_Go(t *testing.T) {
 		assert.Equal(t, uint64(1500), dc.BufferedAmountLowThreshold(), "should match")
 	})
 }
+
+func TestDataChannelBufferedAmount(t *testing.T) {
+	t.Run("set before datachannel becomes open", func(t *testing.T) {
+		report := test.CheckRoutines(t)
+		defer report()
+
+		var nCbs int
+		buf := make([]byte, 1000)
+
+		offerPC, answerPC, err := newPair()
+		if err != nil {
+			t.Fatalf("Failed to create a PC pair for testing")
+		}
+
+		done := make(chan bool)
+
+		answerPC.OnDataChannel(func(d *DataChannel) {
+			// Make sure this is the data channel we were looking for. (Not the one
+			// created in signalPair).
+			if d.Label() != expectedLabel {
+				return
+			}
+			var nPacketsReceived int
+			d.OnMessage(func(msg DataChannelMessage) {
+				nPacketsReceived++
+
+				if nPacketsReceived == 10 {
+					go func() {
+						time.Sleep(time.Second)
+						done <- true
+					}()
+				}
+			})
+			assert.True(t, d.Ordered(), "Ordered should be set to true")
+		})
+
+		dc, err := offerPC.CreateDataChannel(expectedLabel, nil)
+		if err != nil {
+			t.Fatalf("Failed to create a PC pair for testing")
+		}
+
+		assert.True(t, dc.Ordered(), "Ordered should be set to true")
+
+		dc.OnOpen(func() {
+			for i := 0; i < 10; i++ {
+				e := dc.Send(buf)
+				if e != nil {
+					t.Fatalf("Failed to send string on data channel")
+				}
+				assert.Equal(t, uint64(1500), dc.BufferedAmountLowThreshold(), "value mimatch")
+
+				//assert.Equal(t, (i+1)*len(buf), int(dc.BufferedAmount()), "unexpected bufferedAmount")
+			}
+		})
+
+		dc.OnMessage(func(msg DataChannelMessage) {
+		})
+
+		// The value is temporarily stored in the dc object
+		// until the dc gets opened
+		dc.SetBufferedAmountLowThreshold(1500)
+		// The callback function is temporarily stored in the dc object
+		// until the dc gets opened
+		dc.OnBufferedAmountLow(func() {
+			nCbs++
+		})
+
+		err = signalPair(offerPC, answerPC)
+		if err != nil {
+			t.Fatalf("Failed to signal our PC pair for testing")
+		}
+
+		closePair(t, offerPC, answerPC, done)
+
+		assert.True(t, nCbs > 0, "callback should be made at least once")
+	})
+
+	t.Run("set after datachannel becomes open", func(t *testing.T) {
+		report := test.CheckRoutines(t)
+		defer report()
+
+		var nCbs int
+		buf := make([]byte, 1000)
+
+		offerPC, answerPC, err := newPair()
+		if err != nil {
+			t.Fatalf("Failed to create a PC pair for testing")
+		}
+
+		done := make(chan bool)
+
+		answerPC.OnDataChannel(func(d *DataChannel) {
+			// Make sure this is the data channel we were looking for. (Not the one
+			// created in signalPair).
+			if d.Label() != expectedLabel {
+				return
+			}
+			var nPacketsReceived int
+			d.OnMessage(func(msg DataChannelMessage) {
+				nPacketsReceived++
+
+				if nPacketsReceived == 10 {
+					go func() {
+						time.Sleep(time.Second)
+						done <- true
+					}()
+				}
+			})
+			assert.True(t, d.Ordered(), "Ordered should be set to true")
+		})
+
+		dc, err := offerPC.CreateDataChannel(expectedLabel, nil)
+		if err != nil {
+			t.Fatalf("Failed to create a PC pair for testing")
+		}
+
+		assert.True(t, dc.Ordered(), "Ordered should be set to true")
+
+		dc.OnOpen(func() {
+			// The value should directly be passed to sctp
+			dc.SetBufferedAmountLowThreshold(1500)
+			// The callback function should directly be passed to sctp
+			dc.OnBufferedAmountLow(func() {
+				nCbs++
+			})
+
+			for i := 0; i < 10; i++ {
+				e := dc.Send(buf)
+				if e != nil {
+					t.Fatalf("Failed to send string on data channel")
+				}
+				assert.Equal(t, uint64(1500), dc.BufferedAmountLowThreshold(), "value mimatch")
+
+				//assert.Equal(t, (i+1)*len(buf), int(dc.BufferedAmount()), "unexpected bufferedAmount")
+			}
+		})
+
+		dc.OnMessage(func(msg DataChannelMessage) {
+		})
+
+		err = signalPair(offerPC, answerPC)
+		if err != nil {
+			t.Fatalf("Failed to signal our PC pair for testing")
+		}
+
+		closePair(t, offerPC, answerPC, done)
+
+		assert.True(t, nCbs > 0, "callback should be made at least once")
+	})
+}
diff --git a/datachannel_test.go b/datachannel_test.go
index d024232a..c37c4a7d 100644
--- a/datachannel_test.go
+++ b/datachannel_test.go
@@ -106,7 +106,7 @@ func TestDataChannel_Send(t *testing.T) {
 	closePair(t, offerPC, answerPC, done)
 }
 
-func TestDataChannelParamters(t *testing.T) {
+func TestDataChannelParameters(t *testing.T) {
 	report := test.CheckRoutines(t)
 	defer report()