Fixed create data channel

Data channel creation checks if the sctp transport is already
established. Because this check was incorrect, in some cases it returned
an error instead of delaying opening the data channel until the sctp
connection is established. Fixed the check. Added locking to
CreateDataChannel to avoid race condition. Added test.

README.md

@@ -106,6 +106,7 @@ Check out the **[contributing wiki](https://github.com/pion/webrtc/wiki/Contribu
 * [Simonacca Fotokite](https://github.com/simonacca-fotokite)
 * [Marouane](https://github.com/nindolabs) *Fix Offer bundle generation*
 * [Christopher Fry](https://github.com/christopherfry)
+* [Adam Kiss](https://github.com/masterada)
 
 ### License
 MIT License - see [LICENSE](LICENSE) for full text

datachannel.go

@@ -96,6 +96,11 @@ func (api *API) newDataChannel(params *DataChannelParameters, log logging.Levele
 // open opens the datachannel over the sctp transport
 func (d *DataChannel) open(sctpTransport *SCTPTransport) error {
 	d.mu.Lock()
+	if d.sctpTransport != nil {
+		// already open
+		d.mu.Unlock()
+		return nil
+	}
 	d.sctpTransport = sctpTransport
 
 	if err := d.ensureSCTP(); err != nil {

datachannel_test.go

@@ -2,6 +2,7 @@ package webrtc
 
 import (
 	"io"
+	"sync"
 	"testing"
 	"time"
 
@@ -56,54 +57,122 @@ func closeReliabilityParamTest(t *testing.T, pc1, pc2 *PeerConnection, done chan
 }
 
 func TestDataChannel_Send(t *testing.T) {
-	report := test.CheckRoutines(t)
-	defer report()
+	t.Run("before signaling", func(t *testing.T) {
+		report := test.CheckRoutines(t)
+		defer report()
 
-	offerPC, answerPC, err := newPair()
-	if err != nil {
-		t.Fatalf("Failed to create a PC pair for testing")
-	}
+		offerPC, answerPC, err := newPair()
+		if err != nil {
+			t.Fatalf("Failed to create a PC pair for testing")
+		}
 
-	done := make(chan bool)
+		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
+			}
+			d.OnMessage(func(msg DataChannelMessage) {
+				e := d.Send([]byte("Pong"))
+				if e != nil {
+					t.Fatalf("Failed to send string on data channel")
+				}
+			})
+			assert.True(t, d.Ordered(), "Ordered should be set to true")
+		})
 
-	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
+		dc, err := offerPC.CreateDataChannel(expectedLabel, nil)
+		if err != nil {
+			t.Fatalf("Failed to create a PC pair for testing")
 		}
-		d.OnMessage(func(msg DataChannelMessage) {
-			e := d.Send([]byte("Pong"))
+
+		assert.True(t, dc.Ordered(), "Ordered should be set to true")
+
+		dc.OnOpen(func() {
+			e := dc.SendText("Ping")
 			if e != nil {
 				t.Fatalf("Failed to send string on data channel")
 			}
 		})
-		assert.True(t, d.Ordered(), "Ordered should be set to true")
-	})
+		dc.OnMessage(func(msg DataChannelMessage) {
+			done <- true
+		})
 
-	dc, err := offerPC.CreateDataChannel(expectedLabel, nil)
-	if err != nil {
-		t.Fatalf("Failed to create a PC pair for testing")
-	}
+		err = signalPair(offerPC, answerPC)
+		if err != nil {
+			t.Fatalf("Failed to signal our PC pair for testing: %+v", err)
+		}
+
+		closePair(t, offerPC, answerPC, done)
+	})
 
-	assert.True(t, dc.Ordered(), "Ordered should be set to true")
+	t.Run("after connected", func(t *testing.T) {
+		report := test.CheckRoutines(t)
+		defer report()
 
-	dc.OnOpen(func() {
-		e := dc.SendText("Ping")
-		if e != nil {
-			t.Fatalf("Failed to send string on data channel")
+		offerPC, answerPC, err := newPair()
+		if err != nil {
+			t.Fatalf("Failed to create a PC pair for testing")
 		}
-	})
-	dc.OnMessage(func(msg DataChannelMessage) {
-		done <- true
-	})
 
-	err = signalPair(offerPC, answerPC)
-	if err != nil {
-		t.Fatalf("Failed to signal our PC pair for testing")
-	}
+		done := make(chan bool)
 
-	closePair(t, offerPC, answerPC, done)
+		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
+			}
+			d.OnMessage(func(msg DataChannelMessage) {
+				e := d.Send([]byte("Pong"))
+				if e != nil {
+					t.Fatalf("Failed to send string on data channel")
+				}
+			})
+			assert.True(t, d.Ordered(), "Ordered should be set to true")
+		})
+
+		once := &sync.Once{}
+		offerPC.OnICEConnectionStateChange(func(state ICEConnectionState) {
+			if state == ICEConnectionStateConnected || state == ICEConnectionStateCompleted {
+				// wasm fires completed state multiple times
+				once.Do(func() {
+					dc, createErr := offerPC.CreateDataChannel(expectedLabel, nil)
+					if createErr != nil {
+						t.Fatalf("Failed to create a PC pair for testing")
+					}
+
+					assert.True(t, dc.Ordered(), "Ordered should be set to true")
+
+					dc.OnMessage(func(msg DataChannelMessage) {
+						done <- true
+					})
+					// TODO: currently there is no way of properly subscribing to OnOpen with the js binding,
+					// because CreateDataChannel might return an already open data channel
+					//
+					e := dc.SendText("Ping")
+					if e != nil {
+						// wasm binding doesn't fire OnOpen (we probably already missed it)
+						dc.OnOpen(func() {
+							e = dc.SendText("Ping")
+							if e != nil {
+								t.Fatalf("Failed to send string on data channel")
+							}
+						})
+					}
+				})
+			}
+		})
+
+		err = signalPair(offerPC, answerPC)
+		if err != nil {
+			t.Fatalf("Failed to signal our PC pair for testing")
+		}
+
+		closePair(t, offerPC, answerPC, done)
+	})
 }
 
 func TestDataChannelParameters(t *testing.T) {

peerconnection.go

@@ -1056,7 +1056,15 @@ func (pc *PeerConnection) SetRemoteDescription(desc SessionDescription) error {
 
 // openDataChannels opens the existing data channels
 func (pc *PeerConnection) openDataChannels() {
-	for _, d := range pc.dataChannels {
+	pc.mu.Lock()
+	// make a copy of dataChannels to avoid race condition accessing pc.dataChannels
+	dataChannels := make(map[uint16]*DataChannel, len(pc.dataChannels))
+	for k, v := range pc.dataChannels {
+		dataChannels[k] = v
+	}
+	pc.mu.Unlock()
+
+	for _, d := range dataChannels {
 		err := d.open(pc.sctpTransport)
 		if err != nil {
 			pc.log.Warnf("failed to open data channel: %s", err)
@@ -1433,8 +1441,11 @@ func (pc *PeerConnection) AddTransceiver(trackOrKind RTPCodecType, init ...RtpTr
 // and optional DataChannelInit used to configure properties of the
 // underlying channel such as data reliability.
 func (pc *PeerConnection) CreateDataChannel(label string, options *DataChannelInit) (*DataChannel, error) {
+	pc.mu.Lock()
+
 	// https://w3c.github.io/webrtc-pc/#peer-to-peer-data-api (Step #2)
 	if pc.isClosed {
+		pc.mu.Unlock()
 		return nil, &rtcerr.InvalidStateError{Err: ErrConnectionClosed}
 	}
 
@@ -1450,6 +1461,7 @@ func (pc *PeerConnection) CreateDataChannel(label string, options *DataChannelIn
 	if options == nil || options.ID == nil {
 		var err error
 		if params.ID, err = pc.generateDataChannelID(true); err != nil {
+			pc.mu.Unlock()
 			return nil, err
 		}
 	} else {
@@ -1488,19 +1500,25 @@ func (pc *PeerConnection) CreateDataChannel(label string, options *DataChannelIn
 
 	d, err := pc.api.newDataChannel(params, pc.log)
 	if err != nil {
+		pc.mu.Unlock()
 		return nil, err
 	}
 
 	// https://w3c.github.io/webrtc-pc/#peer-to-peer-data-api (Step #16)
 	if d.maxPacketLifeTime != nil && d.maxRetransmits != nil {
+		pc.mu.Unlock()
 		return nil, &rtcerr.TypeError{Err: ErrRetransmitsOrPacketLifeTime}
 	}
 
 	// Remember datachannel
 	pc.dataChannels[params.ID] = d
 
+	sctpReady := pc.sctpTransport != nil && pc.sctpTransport.association != nil
+
+	pc.mu.Unlock()
+
 	// Open if networking already started
-	if pc.sctpTransport != nil {
+	if sctpReady {
 		err = d.open(pc.sctpTransport)
 		if err != nil {
 			return nil, err