Reset state machine after negotiationNeededOp

- Fixes #2774

peerconnection.go

@@ -291,6 +291,16 @@ func (pc *PeerConnection) onNegotiationNeeded() {
 }
 
 func (pc *PeerConnection) negotiationNeededOp() {
+	// non-canon, reset needed state machine and run again if there was a request
+	defer func() {
+		pc.mu.Lock()
+		defer pc.mu.Unlock()
+		if pc.negotiationNeededState == negotiationNeededStateQueue {
+			defer pc.onNegotiationNeeded()
+		}
+		pc.negotiationNeededState = negotiationNeededStateEmpty
+	}()
+
 	// Don't run NegotiatedNeeded checks if OnNegotiationNeeded is not set
 	if handler, ok := pc.onNegotiationNeededHandler.Load().(func()); !ok || handler == nil {
 		return
@@ -307,16 +317,6 @@ func (pc *PeerConnection) negotiationNeededOp() {
 		return
 	}
 
-	// non-canon, run again if there was a request
-	defer func() {
-		pc.mu.Lock()
-		defer pc.mu.Unlock()
-		if pc.negotiationNeededState == negotiationNeededStateQueue {
-			defer pc.onNegotiationNeeded()
-		}
-		pc.negotiationNeededState = negotiationNeededStateEmpty
-	}()
-
 	// Step 2.3
 	if pc.SignalingState() != SignalingStateStable {
 		return

peerconnection_test.go

@@ -553,6 +553,44 @@ func TestNegotiationNeeded(t *testing.T) {
 	assert.NoError(t, pc.Close())
 }
 
+// See https://github.com/pion/webrtc/issues/2774
+func TestNegotiationNeededAddedAfterOpQueueDone(t *testing.T) {
+	lim := test.TimeOut(time.Second * 30)
+	defer lim.Stop()
+
+	report := test.CheckRoutines(t)
+	defer report()
+
+	pc, err := NewPeerConnection(Configuration{})
+	if err != nil {
+		t.Error(err.Error())
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	_, err = pc.CreateDataChannel("initial_data_channel", nil)
+	assert.NoError(t, err)
+
+	// after there are no ops left in the queue, a previously faulty version
+	// of negotiationNeededOp would keep the negotiation needed state in
+	// negotiationNeededStateQueue which will cause all subsequent
+	// onNegotiationNeeded calls to never queue again, only if
+	// OnNegotiationNeeded has not been set yet.
+	for !pc.ops.IsEmpty() {
+		time.Sleep(time.Millisecond)
+	}
+
+	pc.OnNegotiationNeeded(wg.Done)
+
+	_, err = pc.CreateDataChannel("another_data_channel", nil)
+	assert.NoError(t, err)
+
+	wg.Wait()
+
+	assert.NoError(t, pc.Close())
+}
+
 func TestMultipleCreateChannel(t *testing.T) {
 	var wg sync.WaitGroup