Use negotiated data channels in single PC mode (#5017)

* Use negotiated data channels in single PC mode

Bump the protocol version to 18. For protocol >= 18 single peer connection
clients, the server creates its publisher data channels as negotiated with
fixed IDs (100/101/102), so it can send on them without waiting for the
client's DCEP OPEN. Dual PC keeps the in-band channels.

* solve comment

pkg/rtc/transport.go

@@ -72,6 +72,12 @@ const (
 	ReliableDataChannel  = "_reliable"
 	DataTrackDataChannel = "_data_track"
 
+	// Fixed SCTP stream IDs of the publisher data channels which create the data channels as negotiated.
+	// Kept away from low IDs that in-band channels auto-allocate.
+	NegotiatedReliableDataChannelID  uint16 = 100
+	NegotiatedLossyDataChannelID     uint16 = 101
+	NegotiatedDataTrackDataChannelID uint16 = 102
+
 	fastNegotiationFrequency   = 10 * time.Millisecond
 	negotiationFrequency       = 150 * time.Millisecond
 	negotiationFailedTimeout   = 15 * time.Second

pkg/rtc/transportmanager.go

@@ -209,6 +209,11 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro
 		}
 		t.subscriber = subscriber
 	}
+	if t.useNegotiatedDataChannels() {
+		if err := t.createNegotiatedDataChannelsForPublisher(); err != nil {
+			return nil, err
+		}
+	}
 	if !t.params.Migration && t.params.SubscriberAsPrimary {
 		if err := t.createDataChannelsForSubscriber(nil); err != nil {
 			return nil, err
@@ -427,6 +432,46 @@ func (t *TransportManager) handleSendDataResult(err error, kind string, size int
 	return err
 }
 
+func (t *TransportManager) useNegotiatedDataChannels() bool {
+	return t.params.ProtocolVersion.SupportsNegotiatedDataChannels() &&
+		t.params.UseSinglePeerConnection &&
+		!t.params.UseOneShotSignallingMode
+}
+
+func (t *TransportManager) createNegotiatedDataChannelsForPublisher() error {
+	negotiated := true
+
+	reliableID := NegotiatedReliableDataChannelID
+	ordered := true
+	if err := t.publisher.CreateDataChannel(ReliableDataChannel, &webrtc.DataChannelInit{
+		Ordered:    &ordered,
+		ID:         &reliableID,
+		Negotiated: &negotiated,
+	}); err != nil {
+		return err
+	}
+
+	ordered = false
+	retransmits := uint16(0)
+	lossyID := NegotiatedLossyDataChannelID
+	if err := t.publisher.CreateDataChannel(LossyDataChannel, &webrtc.DataChannelInit{
+		Ordered:        &ordered,
+		MaxRetransmits: &retransmits,
+		ID:             &lossyID,
+		Negotiated:     &negotiated,
+	}); err != nil {
+		return err
+	}
+
+	dataTrackID := NegotiatedDataTrackDataChannelID
+	return t.publisher.CreateDataChannel(DataTrackDataChannel, &webrtc.DataChannelInit{
+		Ordered:        &ordered,
+		MaxRetransmits: &retransmits,
+		ID:             &dataTrackID,
+		Negotiated:     &negotiated,
+	})
+}
+
 func (t *TransportManager) createDataChannelsForSubscriber(pendingDataChannels []*livekit.DataChannelInfo) error {
 	var (
 		reliableID, lossyID, dataTrackID          uint16
@@ -921,6 +966,10 @@ func (t *TransportManager) ProcessPendingPublisherDataChannels() {
 	t.pendingDataChannelsPublisher = nil
 	t.lock.Unlock()
 
+	if t.useNegotiatedDataChannels() {
+		return
+	}
+
 	ordered := true
 	negotiated := true
 

pkg/rtc/types/protocol_version.go

@@ -16,7 +16,7 @@ package types
 
 type ProtocolVersion int
 
-const CurrentProtocol = 17
+const CurrentProtocol = 18
 
 func (v ProtocolVersion) SupportsPackedStreamId() bool {
 	return v > 0
@@ -103,3 +103,7 @@ func (v ProtocolVersion) SupportsMoving() bool {
 func (v ProtocolVersion) SupportsPacketTrailer() bool {
 	return v > 16
 }
+
+func (v ProtocolVersion) SupportsNegotiatedDataChannels() bool {
+	return v > 17
+}

test/client/client.go

@@ -359,35 +359,39 @@ func (c *RTCClient) createTransport(rtcconf webrtc.Configuration) error {
 	})
 
 	ordered := true
-	if err := c.publisher.CreateDataChannel(rtc.ReliableDataChannel, &webrtc.DataChannelInit{
-		Ordered: &ordered,
-	}); err != nil {
-		return err
+	if c.useSinglePeerConnection {
+		if err := createNegotiatedDataChannels(c.publisher); err != nil {
+			return err
+		}
+	} else {
+		if err := c.publisher.CreateDataChannel(rtc.ReliableDataChannel, &webrtc.DataChannelInit{
+			Ordered: &ordered,
+		}); err != nil {
+			return err
+		}
 	}
-
 	if err := c.publisher.CreateDataChannel("pubraw", &webrtc.DataChannelInit{
 		Ordered: &ordered,
 	}); err != nil {
 		return err
 	}
 
-	ordered = false
-	maxRetransmits := uint16(0)
-	if err := c.publisher.CreateDataChannel(rtc.LossyDataChannel, &webrtc.DataChannelInit{
-		Ordered:        &ordered,
-		MaxRetransmits: &maxRetransmits,
-	}); err != nil {
-		return err
-	}
-
-	if err := c.publisher.CreateDataChannel(rtc.DataTrackDataChannel, &webrtc.DataChannelInit{
-		Ordered:        &ordered,
-		MaxRetransmits: &maxRetransmits,
-	}); err != nil {
-		return err
-	}
-
 	if !c.useSinglePeerConnection {
+		maxRetransmits := uint16(0)
+		ordered = false
+		if err := c.publisher.CreateDataChannel(rtc.LossyDataChannel, &webrtc.DataChannelInit{
+			Ordered:        &ordered,
+			MaxRetransmits: &maxRetransmits,
+		}); err != nil {
+			return err
+		}
+		if err := c.publisher.CreateDataChannel(rtc.DataTrackDataChannel, &webrtc.DataChannelInit{
+			Ordered:        &ordered,
+			MaxRetransmits: &maxRetransmits,
+		}); err != nil {
+			return err
+		}
+
 		subscriberHandler := &transportfakes.FakeHandler{}
 		c.subscriber, err = rtc.NewPCTransport(rtc.TransportParams{
 			Config:                           &conf,
@@ -1405,3 +1409,38 @@ func (c *RTCClient) SendNacks(count int) {
 func encodeQueryParam(key, value string) string {
 	return fmt.Sprintf("&%s=%s", url.QueryEscape(key), url.QueryEscape(value))
 }
+
+// createNegotiatedDataChannels creates the well-known data channels the way protocol >= 18 clients do:
+// out-of-band with fixed IDs, matching what the server creates in single peer connection mode.
+func createNegotiatedDataChannels(t *rtc.PCTransport) error {
+	negotiated := true
+	ordered := true
+	reliableID := rtc.NegotiatedReliableDataChannelID
+	if err := t.CreateDataChannel(rtc.ReliableDataChannel, &webrtc.DataChannelInit{
+		Ordered:    &ordered,
+		Negotiated: &negotiated,
+		ID:         &reliableID,
+	}); err != nil {
+		return err
+	}
+
+	ordered = false
+	maxRetransmits := uint16(0)
+	lossyID := rtc.NegotiatedLossyDataChannelID
+	if err := t.CreateDataChannel(rtc.LossyDataChannel, &webrtc.DataChannelInit{
+		Ordered:        &ordered,
+		MaxRetransmits: &maxRetransmits,
+		Negotiated:     &negotiated,
+		ID:             &lossyID,
+	}); err != nil {
+		return err
+	}
+
+	dataTrackID := rtc.NegotiatedDataTrackDataChannelID
+	return t.CreateDataChannel(rtc.DataTrackDataChannel, &webrtc.DataChannelInit{
+		Ordered:        &ordered,
+		MaxRetransmits: &maxRetransmits,
+		Negotiated:     &negotiated,
+		ID:             &dataTrackID,
+	})
+}