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

Embedded turn test (#4412)

Browse Source 
* Embedded turn test

* unused code

* fix test
This commit is contained in:
cnderrauber
2026-03-30 22:24:00 +08:00
committed by GitHub
parent 563266548b
commit 397cd09a90
2 changed files with 129 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/client/client.go
+94 -26
View File
@@ -63,6 +63,9 @@ type RTCClient struct {
conn *websocket.Conn
publisher *rtc.PCTransport
subscriber *rtc.PCTransport
enabledCodecs []*livekit.Codec
forceRelay bool
transportReady chan struct{}
// sid => track
localTracks map[string]webrtc.TrackLocal
trackSenders map[string]*webrtc.RTPSender
@@ -137,6 +140,7 @@ type Options struct {
SignalResponseInterceptor SignalResponseInterceptor
UseJoinRequestQueryParam bool
RTCServicePath string
ForceRelay bool
}
func NewWebSocketConn(host, token string, opts *Options) (*websocket.Conn, error) {
@@ -223,8 +227,6 @@ func SetAuthorizationToken(header http.Header, token string) {
}
func NewRTCClient(conn *websocket.Conn, useSinglePeerConnection bool, opts *Options) (*RTCClient, error) {
var err error
c := &RTCClient{
useSinglePeerConnection: useSinglePeerConnection,
conn: conn,
@@ -238,19 +240,11 @@ func NewRTCClient(conn *websocket.Conn, useSinglePeerConnection bool, opts *Opti
bytesReceived: make(map[livekit.ParticipantID]uint64),
pendingPublishedDataTracks: make(map[uint16]*livekit.DataTrackInfo),
subscribedDataTracks: make(map[livekit.ParticipantID]map[uint16]*DataTrackRemote),
transportReady: make(chan struct{}),
}
c.nextDataTrackHandle.Store(uint32(rand.IntN(8192)))
c.ctx, c.cancel = context.WithCancel(context.Background())
conf := rtc.WebRTCConfig{
WebRTCConfig: rtcconfig.WebRTCConfig{
Configuration: rtcConf,
},
}
conf.SettingEngine.SetLite(false)
conf.SettingEngine.SetAnsweringDTLSRole(webrtc.DTLSRoleClient)
ff := buffer.NewFactoryOfBufferFactory(500, 200)
conf.SetBufferFactory(ff.CreateBufferFactory())
var codecs []*livekit.Codec
for _, codec := range []*livekit.Codec{
{
@@ -276,6 +270,29 @@ func NewRTCClient(conn *websocket.Conn, useSinglePeerConnection bool, opts *Opti
codecs = append(codecs, codec)
}
}
c.enabledCodecs = codecs
if opts != nil {
c.signalRequestInterceptor = opts.SignalRequestInterceptor
c.signalResponseInterceptor = opts.SignalResponseInterceptor
c.forceRelay = opts.ForceRelay
}
return c, nil
}
func (c *RTCClient) createTransport(rtcconf webrtc.Configuration) error {
var err error
conf := rtc.WebRTCConfig{
WebRTCConfig: rtcconfig.WebRTCConfig{
Configuration: rtcconf,
},
}
conf.SettingEngine.SetLite(false)
conf.SettingEngine.SetAnsweringDTLSRole(webrtc.DTLSRoleClient)
ff := buffer.NewFactoryOfBufferFactory(500, 200)
conf.SetBufferFactory(ff.CreateBufferFactory())
//
// The signal targets are from point of view of server.
@@ -287,7 +304,7 @@ func NewRTCClient(conn *websocket.Conn, useSinglePeerConnection bool, opts *Opti
c.publisher, err = rtc.NewPCTransport(rtc.TransportParams{
Config: &conf,
DirectionConfig: conf.Subscriber,
EnabledCodecs: codecs,
EnabledCodecs: c.enabledCodecs,
IsOfferer: true,
IsSendSide: true,
Handler: publisherHandler,
@@ -297,7 +314,7 @@ func NewRTCClient(conn *websocket.Conn, useSinglePeerConnection bool, opts *Opti
EnableDataTracks: true,
})
if err != nil {
return nil, err
return err
}
publisherHandler.OnICECandidateCalls(func(ic *webrtc.ICECandidate, t livekit.SignalTarget) error {
@@ -342,13 +359,13 @@ func NewRTCClient(conn *websocket.Conn, useSinglePeerConnection bool, opts *Opti
if err := c.publisher.CreateDataChannel(rtc.ReliableDataChannel, &webrtc.DataChannelInit{
Ordered: &ordered,
}); err != nil {
return nil, err
return err
}
if err := c.publisher.CreateDataChannel("pubraw", &webrtc.DataChannelInit{
Ordered: &ordered,
}); err != nil {
return nil, err
return err
}
ordered = false
@@ -357,14 +374,14 @@ func NewRTCClient(conn *websocket.Conn, useSinglePeerConnection bool, opts *Opti
Ordered: &ordered,
MaxRetransmits: &maxRetransmits,
}); err != nil {
return nil, err
return err
}
if err := c.publisher.CreateDataChannel(rtc.DataTrackDataChannel, &webrtc.DataChannelInit{
Ordered: &ordered,
MaxRetransmits: &maxRetransmits,
}); err != nil {
return nil, err
return err
}
if !c.useSinglePeerConnection {
@@ -372,7 +389,7 @@ func NewRTCClient(conn *websocket.Conn, useSinglePeerConnection bool, opts *Opti
c.subscriber, err = rtc.NewPCTransport(rtc.TransportParams{
Config: &conf,
DirectionConfig: conf.Publisher,
EnabledCodecs: codecs,
EnabledCodecs: c.enabledCodecs,
Handler: subscriberHandler,
DatachannelMaxReceiverBufferSize: 1500,
DatachannelSlowThreshold: 1024 * 1024 * 1024,
@@ -380,14 +397,14 @@ func NewRTCClient(conn *websocket.Conn, useSinglePeerConnection bool, opts *Opti
EnableDataTracks: true,
})
if err != nil {
return nil, err
return err
}
ordered := true
if err := c.subscriber.CreateReadableDataChannel("subraw", &webrtc.DataChannelInit{
Ordered: &ordered,
}); err != nil {
return nil, err
return err
}
subscriberHandler.OnICECandidateCalls(func(ic *webrtc.ICECandidate, t livekit.SignalTarget) error {
@@ -446,12 +463,7 @@ func NewRTCClient(conn *websocket.Conn, useSinglePeerConnection bool, opts *Opti
go c.ensurePublisherConnected()
}
if opts != nil {
c.signalRequestInterceptor = opts.SignalRequestInterceptor
c.signalResponseInterceptor = opts.SignalResponseInterceptor
}
return c, nil
return nil
}
func (c *RTCClient) ID() livekit.ParticipantID {
@@ -497,6 +509,29 @@ func (c *RTCClient) handleSignalResponse(res *livekit.SignalResponse) error {
c.remoteParticipants[livekit.ParticipantID(p.Sid)] = p
}
c.lock.Unlock()
var iceServers []webrtc.ICEServer
for _, is := range msg.Join.IceServers {
iceServers = append(iceServers, webrtc.ICEServer{
URLs: is.Urls,
Username: is.Username,
Credential: is.Credential,
})
}
if len(iceServers) == 0 {
iceServers = rtcConf.ICEServers
}
rtcconf := rtcConf
rtcconf.ICEServers = iceServers
if c.forceRelay {
rtcconf.ICETransportPolicy = webrtc.ICETransportPolicyRelay
}
if err := c.createTransport(rtcconf); err != nil {
return err
}
close(c.transportReady)
// if publish only, negotiate
if !msg.Join.SubscriberPrimary {
c.subscriberAsPrimary.Store(false)
@@ -846,6 +881,11 @@ func AddTrackNoWriter() AddTrackOption {
}
func (c *RTCClient) AddTrack(track *webrtc.TrackLocalStaticSample, path string, opts ...AddTrackOption) (writer TrackWriter, err error) {
select {
case <-c.transportReady:
case <-c.ctx.Done():
return nil, c.ctx.Err()
}
var params AddTrackParams
for _, opt := range opts {
opt(&params)
@@ -1062,6 +1102,12 @@ func (c *RTCClient) LastAnswer() *webrtc.SessionDescription {
}
func (c *RTCClient) ensurePublisherConnected() error {
select {
case <-c.transportReady:
case <-c.ctx.Done():
return c.ctx.Err()
}
if c.publisher.HasEverConnected() {
return nil
}
@@ -1294,6 +1340,28 @@ func (c *RTCClient) BytesReceived() uint64 {
return total
}
func (c *RTCClient) IsLocalCandidateRelaySelected() bool {
var info *types.ICEConnectionInfo
if c.subscriberAsPrimary.Load() {
if c.subscriber != nil {
info = c.subscriber.GetICEConnectionInfo()
}
} else {
if c.publisher != nil {
info = c.publisher.GetICEConnectionInfo()
}
}
if info == nil {
return false
}
for _, local := range info.Local {
if local.SelectedOrder > 0 && local.Local != nil && local.Local.Typ == webrtc.ICECandidateTypeRelay {
return true
}
}
return false
}
func (c *RTCClient) SendNacks(count int) {
var packets []rtcp.Packet
c.lock.Lock()
test/singlenode_test.go
+35
View File
@@ -1100,3 +1100,38 @@ func TestSinglePublisherDataTrack(t *testing.T) {
})
}
}
func TestTurnRelay(t *testing.T) {
if testing.Short() {
t.SkipNow()
return
}
s := createSingleNodeServer(func(c *config.Config) {
c.TURN.Enabled = true
c.TURN.UDPPort = 3478
})
go func() {
if err := s.Start(); err != nil {
logger.Errorw("server returned error", err)
}
}()
defer s.Stop(true)
waitForServerToStart(s)
c1 := createRTCClient("relay_c1", defaultServerPort, testRTCServicePathv0, &testclient.Options{
AutoSubscribe: true,
ForceRelay: true,
})
defer c1.Stop()
waitUntilConnected(t, c1)
testutils.WithTimeout(t, func() string {
if !c1.IsLocalCandidateRelaySelected() {
return "expected local candidate to be relay"
}
return ""
})
}