diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index df97dd4b6..4f98d6028 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -178,20 +178,21 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro if !t.params.UseOneShotSignallingMode { lgr := LoggerWithPCTarget(params.Logger, livekit.SignalTarget_SUBSCRIBER) subscriber, err := NewPCTransport(TransportParams{ - ProtocolVersion: params.ProtocolVersion, - Config: params.Config, - DirectionConfig: params.Config.Subscriber, - CongestionControlConfig: params.CongestionControlConfig, - EnabledCodecs: params.EnabledSubscribeCodecs, - Logger: lgr, - ClientInfo: params.ClientInfo, - IsOfferer: true, - IsSendSide: true, - AllowPlayoutDelay: params.AllowPlayoutDelay, - DatachannelSlowThreshold: params.DatachannelSlowThreshold, - Transport: livekit.SignalTarget_SUBSCRIBER, - Handler: TransportManagerTransportHandler{params.SubscriberHandler, t, lgr}, - FireOnTrackBySdp: params.FireOnTrackBySdp, + ProtocolVersion: params.ProtocolVersion, + Config: params.Config, + DirectionConfig: params.Config.Subscriber, + CongestionControlConfig: params.CongestionControlConfig, + EnabledCodecs: params.EnabledSubscribeCodecs, + Logger: lgr, + ClientInfo: params.ClientInfo, + IsOfferer: true, + IsSendSide: true, + AllowPlayoutDelay: params.AllowPlayoutDelay, + DataChannelMaxBufferedAmount: params.DataChannelMaxBufferedAmount, + DatachannelSlowThreshold: params.DatachannelSlowThreshold, + Transport: livekit.SignalTarget_SUBSCRIBER, + Handler: TransportManagerTransportHandler{params.SubscriberHandler, t, lgr}, + FireOnTrackBySdp: params.FireOnTrackBySdp, }) if err != nil { return nil, err diff --git a/test/client/client.go b/test/client/client.go index e856c2ac6..13cd6e84b 100644 --- a/test/client/client.go +++ b/test/client/client.go @@ -261,19 +261,53 @@ func NewRTCClient(conn *websocket.Conn, opts *Options) (*RTCClient, error) { // i. e. the publisher transport on client side has SUBSCRIBER signal target (i. e. publisher is offerer). // Same applies for subscriber transport also // - publisherHandler := &transportfakes.FakeHandler{} - c.publisher, err = rtc.NewPCTransport(rtc.TransportParams{ - Config: &conf, - DirectionConfig: conf.Subscriber, - EnabledCodecs: codecs, - IsOfferer: true, - IsSendSide: true, - Handler: publisherHandler, - DatachannelSlowThreshold: 1024 * 1024 * 1024, - }) - if err != nil { - return nil, err + if !c.protocolVersion.SupportsSinglePeerConnection() { + publisherHandler := &transportfakes.FakeHandler{} + c.publisher, err = rtc.NewPCTransport(rtc.TransportParams{ + Config: &conf, + DirectionConfig: conf.Subscriber, + EnabledCodecs: codecs, + IsOfferer: true, + IsSendSide: true, + Handler: publisherHandler, + DatachannelSlowThreshold: 1024 * 1024 * 1024, + }) + if err != nil { + return nil, err + } + + publisherHandler.OnICECandidateCalls(func(ic *webrtc.ICECandidate, t livekit.SignalTarget) error { + return c.SendIceCandidate(ic, livekit.SignalTarget_PUBLISHER) + }) + publisherHandler.OnOfferCalls(c.onOffer) + publisherHandler.OnFullyEstablishedCalls(func() { + logger.Debugw("publisher fully established", "participant", c.localParticipant.Identity, "pID", c.localParticipant.Sid) + c.publisherFullyEstablished.Store(true) + }) + + ordered := true + if err := c.publisher.CreateDataChannel(rtc.ReliableDataChannel, &webrtc.DataChannelInit{ + Ordered: &ordered, + }); err != nil { + return nil, err + } + + ordered = false + maxRetransmits := uint16(0) + if err := c.publisher.CreateDataChannel(rtc.LossyDataChannel, &webrtc.DataChannelInit{ + Ordered: &ordered, + MaxRetransmits: &maxRetransmits, + }); err != nil { + return nil, err + } + + if err := c.publisher.CreateDataChannel("pubraw", &webrtc.DataChannelInit{ + Ordered: &ordered, + }); err != nil { + return nil, err + } } + subscriberHandler := &transportfakes.FakeHandler{} c.subscriber, err = rtc.NewPCTransport(rtc.TransportParams{ Config: &conf, @@ -281,43 +315,14 @@ func NewRTCClient(conn *websocket.Conn, opts *Options) (*RTCClient, error) { EnabledCodecs: codecs, Handler: subscriberHandler, DatachannelMaxReceiverBufferSize: 1500, + DatachannelSlowThreshold: 1024 * 1024 * 1024, FireOnTrackBySdp: true, }) if err != nil { return nil, err } - publisherHandler.OnICECandidateCalls(func(ic *webrtc.ICECandidate, t livekit.SignalTarget) error { - return c.SendIceCandidate(ic, livekit.SignalTarget_PUBLISHER) - }) - publisherHandler.OnOfferCalls(c.onOffer) - publisherHandler.OnFullyEstablishedCalls(func() { - logger.Debugw("publisher fully established", "participant", c.localParticipant.Identity, "pID", c.localParticipant.Sid) - c.publisherFullyEstablished.Store(true) - }) - - ordered := true - if err := c.publisher.CreateDataChannel(rtc.ReliableDataChannel, &webrtc.DataChannelInit{ - Ordered: &ordered, - }); err != nil { - return nil, err - } - - ordered = false - maxRetransmits := uint16(0) - if err := c.publisher.CreateDataChannel(rtc.LossyDataChannel, &webrtc.DataChannelInit{ - Ordered: &ordered, - MaxRetransmits: &maxRetransmits, - }); err != nil { - return nil, err - } - - if err := c.publisher.CreateDataChannel("pubraw", &webrtc.DataChannelInit{ - Ordered: &ordered, - }); err != nil { - return nil, err - } - + ordered := false if err := c.subscriber.CreateReadableDataChannel("subraw", &webrtc.DataChannelInit{ Ordered: &ordered, }); err != nil { @@ -424,7 +429,9 @@ func (c *RTCClient) handleSignalResponse(res *livekit.SignalResponse) error { // if publish only, negotiate if !msg.Join.SubscriberPrimary { c.subscriberAsPrimary.Store(false) - c.publisher.Negotiate(false) + if !c.protocolVersion.SupportsSinglePeerConnection() { + c.publisher.Negotiate(false) + } } else { c.subscriberAsPrimary.Store(true) } @@ -480,12 +487,14 @@ func (c *RTCClient) handleSignalResponse(res *livekit.SignalResponse) error { case *livekit.SignalResponse_TrackUnpublished: sid := msg.TrackUnpublished.TrackSid c.lock.Lock() - sender := c.trackSenders[sid] - if sender != nil { - if err := c.publisher.RemoveTrack(sender); err != nil { - logger.Errorw("Could not unpublish track", err) + if !c.protocolVersion.SupportsSinglePeerConnection() { + sender := c.trackSenders[sid] + if sender != nil { + if err := c.publisher.RemoveTrack(sender); err != nil { + logger.Errorw("Could not unpublish track", err) + } + c.publisher.Negotiate(false) } - c.publisher.Negotiate(false) } delete(c.trackSenders, sid) delete(c.localTracks, sid) @@ -510,7 +519,7 @@ func (c *RTCClient) WaitUntilConnected() error { } return fmt.Errorf("%s could not connect after timeout", id) case <-time.After(10 * time.Millisecond): - if c.subscriberAsPrimary.Load() { + if c.subscriberAsPrimary.Load() || c.protocolVersion.SupportsSinglePeerConnection() { if c.subscriberFullyEstablished.Load() { return nil } @@ -586,7 +595,9 @@ func (c *RTCClient) Stop() { c.publisherFullyEstablished.Store(false) c.subscriberFullyEstablished.Store(false) _ = c.conn.Close() - c.publisher.Close() + if !c.protocolVersion.SupportsSinglePeerConnection() { + c.publisher.Close() + } c.subscriber.Close() c.cancel() } @@ -656,7 +667,7 @@ func (c *RTCClient) SetAttributes(attrs map[string]string) error { } func (c *RTCClient) hasPrimaryEverConnected() bool { - if c.subscriberAsPrimary.Load() { + if c.subscriberAsPrimary.Load() || c.protocolVersion.SupportsSinglePeerConnection() { return c.subscriber.HasEverConnected() } else { return c.publisher.HasEverConnected() @@ -852,11 +863,15 @@ func (c *RTCClient) PublishData(data []byte, kind livekit.DataPacket_Kind) error } func (c *RTCClient) PublishDataUnlabeled(data []byte) error { - if err := c.ensurePublisherConnected(); err != nil { - return err + if !c.protocolVersion.SupportsSinglePeerConnection() { + if err := c.ensurePublisherConnected(); err != nil { + return err + } + + return c.publisher.SendDataMessageUnlabeled(data, true, "test") } - return c.publisher.SendDataMessageUnlabeled(data, true, "test") + return c.subscriber.SendDataMessageUnlabeled(data, true, "test") } func (c *RTCClient) GetPublishedTrackIDs() []string { diff --git a/test/singlenode_test.go b/test/singlenode_test.go index 64aa4ad93..6c9350d3c 100644 --- a/test/singlenode_test.go +++ b/test/singlenode_test.go @@ -495,6 +495,7 @@ func TestSingleNodeUpdateSubscriptionPermissions(t *testing.T) { defer finish() pub := createRTCClient("pub", defaultServerPort, nil) + grant := &auth.VideoGrant{RoomJoin: true, Room: testRoom} grant.SetCanSubscribe(false) at := auth.NewAccessToken(testApiKey, testApiSecret). @@ -955,6 +956,7 @@ func TestFireTrackBySdp(t *testing.T) { } } +// SINGLE-PEER-CONNECTION-TODO: delete this test and make all other tests for both two peer connections and one peer connection func TestSinglePeerConnection(t *testing.T) { _, finish := setupSingleNodeTest("TestSinglePeerConnection") defer finish()