Repository: pion/webrtc (go)

Issue:
SetBufferedAmountLowThreshold() and OnBufferedAmountLow() calls are ignored during OnDataChannel callback
### Your environment.
  - Version: v3.1.58
  - Browser: not relevant

### What did you do?
While I was testing data channel's flow control feature, I noticed values set with these calls during the OnDataChannel callback were ignored (were not effective). Doing so inside OnOpen() work fine.


### What did you expect?
Calls to SetBufferedAmountLowThreshold() and OnBufferedAmountLow() (before OnOpen) should be effective.



Pull request description:
Fix ignored calls during OnDataChannel
Set cached values passed by SetBufferedAmountLowThreshold and OnBufferedAmountLow correctly when the underlying datachannel become open.

#### Description
These method calls are ignored when they are called during OnDataChannel callback, before OnOpen callback.
* dc.SetBufferedAmountLowThreshold()
* dc.OnBufferedAmountLow()

The values are cached on webrtc.DataChannel object, but these were not passed when OnDataChannel is called.

#### Reference issue
Fixes #2441


Patch:
diff --git a/datachannel.go b/datachannel.go
index 6e41fdc1..1ee1516c 100644
--- a/datachannel.go
+++ b/datachannel.go
@@ -307,6 +307,9 @@ func (d *DataChannel) handleOpen(dc *datachannel.DataChannel, isRemote, isAlread
 	// * remote datachannels should fire OnOpened. This isn't spec compliant, but we can't break behavior yet
 	// * already negotiated datachannels should fire OnOpened
 	if d.api.settingEngine.detach.DataChannels || isRemote || isAlreadyNegotiated {
+		// bufferedAmountLowThreshold and onBufferedAmountLow might be set earlier
+		d.dataChannel.SetBufferedAmountLowThreshold(d.bufferedAmountLowThreshold)
+		d.dataChannel.OnBufferedAmountLow(d.onBufferedAmountLow)
 		d.onOpen()
 	} else {
 		dc.OnOpen(func() {


Test patch:
diff --git a/datachannel_go_test.go b/datachannel_go_test.go
index 5c9036f1..9729967a 100644
--- a/datachannel_go_test.go
+++ b/datachannel_go_test.go
@@ -14,6 +14,7 @@ import (
 	"regexp"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -182,7 +183,11 @@ func TestDataChannelBufferedAmount(t *testing.T) {
 		report := test.CheckRoutines(t)
 		defer report()
 
-		var nCbs int
+		var nOfferBufferedAmountLowCbs uint32
+		var offerBufferedAmountLowThreshold uint64 = 1500
+		var nAnswerBufferedAmountLowCbs uint32
+		var answerBufferedAmountLowThreshold uint64 = 1400
+
 		buf := make([]byte, 1000)
 
 		offerPC, answerPC, err := newPair()
@@ -190,56 +195,81 @@ func TestDataChannelBufferedAmount(t *testing.T) {
 			t.Fatalf("Failed to create a PC pair for testing")
 		}
 
+		nPacketsToSend := int(10)
+		var nOfferReceived uint32
+		var nAnswerReceived uint32
+
 		done := make(chan bool)
 
-		answerPC.OnDataChannel(func(d *DataChannel) {
+		answerPC.OnDataChannel(func(answerDC *DataChannel) {
 			// Make sure this is the data channel we were looking for. (Not the one
 			// created in signalPair).
-			if d.Label() != expectedLabel {
+			if answerDC.Label() != expectedLabel {
 				return
 			}
-			var nPacketsReceived int
-			d.OnMessage(func(msg DataChannelMessage) {
-				nPacketsReceived++
 
-				if nPacketsReceived == 10 {
-					go func() {
-						time.Sleep(time.Second)
-						done <- true
-					}()
+			answerDC.OnOpen(func() {
+				assert.Equal(t, answerBufferedAmountLowThreshold, answerDC.BufferedAmountLowThreshold(), "value mismatch")
+
+				for i := 0; i < nPacketsToSend; i++ {
+					e := answerDC.Send(buf)
+					if e != nil {
+						t.Fatalf("Failed to send string on data channel")
+					}
+				}
+			})
+
+			answerDC.OnMessage(func(msg DataChannelMessage) {
+				atomic.AddUint32(&nAnswerReceived, 1)
+			})
+			assert.True(t, answerDC.Ordered(), "Ordered should be set to true")
+
+			// The value is temporarily stored in the answerDC object
+			// until the answerDC gets opened
+			answerDC.SetBufferedAmountLowThreshold(answerBufferedAmountLowThreshold)
+			// The callback function is temporarily stored in the answerDC object
+			// until the answerDC gets opened
+			answerDC.OnBufferedAmountLow(func() {
+				atomic.AddUint32(&nAnswerBufferedAmountLowCbs, 1)
+				if atomic.LoadUint32(&nOfferBufferedAmountLowCbs) > 0 {
+					done <- true
 				}
 			})
-			assert.True(t, d.Ordered(), "Ordered should be set to true")
 		})
 
-		dc, err := offerPC.CreateDataChannel(expectedLabel, nil)
+		offerDC, 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")
+		assert.True(t, offerDC.Ordered(), "Ordered should be set to true")
 
-		dc.OnOpen(func() {
-			for i := 0; i < 10; i++ {
-				e := dc.Send(buf)
+		offerDC.OnOpen(func() {
+			assert.Equal(t, offerBufferedAmountLowThreshold, offerDC.BufferedAmountLowThreshold(), "value mismatch")
+
+			for i := 0; i < nPacketsToSend; i++ {
+				e := offerDC.Send(buf)
 				if e != nil {
 					t.Fatalf("Failed to send string on data channel")
 				}
-				assert.Equal(t, uint64(1500), dc.BufferedAmountLowThreshold(), "value mismatch")
-				// assert.Equal(t, (i+1)*len(buf), int(dc.BufferedAmount()), "unexpected bufferedAmount")
+				// assert.Equal(t, (i+1)*len(buf), int(offerDC.BufferedAmount()), "unexpected bufferedAmount")
 			}
 		})
 
-		dc.OnMessage(func(msg DataChannelMessage) {
+		offerDC.OnMessage(func(msg DataChannelMessage) {
+			atomic.AddUint32(&nOfferReceived, 1)
 		})
 
-		// 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++
+		// The value is temporarily stored in the offerDC object
+		// until the offerDC gets opened
+		offerDC.SetBufferedAmountLowThreshold(offerBufferedAmountLowThreshold)
+		// The callback function is temporarily stored in the offerDC object
+		// until the offerDC gets opened
+		offerDC.OnBufferedAmountLow(func() {
+			atomic.AddUint32(&nOfferBufferedAmountLowCbs, 1)
+			if atomic.LoadUint32(&nAnswerBufferedAmountLowCbs) > 0 {
+				done <- true
+			}
 		})
 
 		err = signalPair(offerPC, answerPC)
@@ -249,7 +279,10 @@ func TestDataChannelBufferedAmount(t *testing.T) {
 
 		closePair(t, offerPC, answerPC, done)
 
-		assert.True(t, nCbs > 0, "callback should be made at least once")
+		t.Logf("nOfferBufferedAmountLowCbs : %d", nOfferBufferedAmountLowCbs)
+		t.Logf("nAnswerBufferedAmountLowCbs: %d", nAnswerBufferedAmountLowCbs)
+		assert.True(t, nOfferBufferedAmountLowCbs > uint32(0), "callback should be made at least once")
+		assert.True(t, nAnswerBufferedAmountLowCbs > uint32(0), "callback should be made at least once")
 	})
 
 	t.Run("set after datachannel becomes open", func(t *testing.T) {