diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index c2db48f92..2a2fe2981 100644 --- a/pkg/rtc/transport.go +++ b/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 diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index e6d983a43..afdfd5991 100644 --- a/pkg/rtc/transportmanager.go +++ b/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 diff --git a/pkg/rtc/types/protocol_version.go b/pkg/rtc/types/protocol_version.go index 9edab8b07..9b8d05be1 100644 --- a/pkg/rtc/types/protocol_version.go +++ b/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 +} diff --git a/test/client/client.go b/test/client/client.go index 14fd108a2..aef43b857 100644 --- a/test/client/client.go +++ b/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, + }) +}