From 397cd09a903d4245652aa53094c0737be73f24d9 Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Mon, 30 Mar 2026 22:24:00 +0800 Subject: [PATCH] Embedded turn test (#4412) * Embedded turn test * unused code * fix test --- test/client/client.go | 120 +++++++++++++++++++++++++++++++--------- test/singlenode_test.go | 35 ++++++++++++ 2 files changed, 129 insertions(+), 26 deletions(-) diff --git a/test/client/client.go b/test/client/client.go index ac9487f6f..00b49012f 100644 --- a/test/client/client.go +++ b/test/client/client.go @@ -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(¶ms) @@ -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() diff --git a/test/singlenode_test.go b/test/singlenode_test.go index 3808c053a..d1cbf152c 100644 --- a/test/singlenode_test.go +++ b/test/singlenode_test.go @@ -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 "" + }) +}