dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-28 20:48:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

WIP

Browse Source
This commit is contained in:
boks1971
2025-08-15 02:16:15 +05:30
parent 94b4a72b49
commit 3842882a04
3 changed files with 87 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/transportmanager.go
+15 -14
View File
@@ -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
test/client/client.go
+70 -55
View File
@@ -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 {
test/singlenode_test.go
+2
View File
@@ -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()