From 0d4f5b08699a6e12200bd50ba975a3edeaaa0bf3 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Fri, 17 Jul 2026 12:13:41 -0700 Subject: [PATCH] Update tests --- test/client/client.go | 25 +++++++------------------ test/singlenode_test.go | 38 ++++++++++++++------------------------ 2 files changed, 21 insertions(+), 42 deletions(-) diff --git a/test/client/client.go b/test/client/client.go index d062a34df..10b43531c 100644 --- a/test/client/client.go +++ b/test/client/client.go @@ -110,8 +110,8 @@ type RTCClient struct { pendingPublishedDataTracks map[uint16]*livekit.DataTrackInfo pendingDataTrackWriters []TrackWriter subscribedDataTracks map[livekit.ParticipantID]map[uint16]*DataTrackRemote - // last received subscriber count per published data track handle - dataTrackDemands map[uint16]uint32 + // whether each data track currently has subscribers + dataTrackDemands map[uint16]bool } var ( @@ -242,7 +242,7 @@ 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), - dataTrackDemands: make(map[uint16]uint32), + dataTrackDemands: make(map[uint16]bool), transportReady: make(chan struct{}), } c.nextDataTrackHandle.Store(uint32(rand.IntN(8192))) @@ -696,10 +696,10 @@ func (c *RTCClient) handleSignalResponse(res *livekit.SignalResponse) error { "received data track demand update", "participant", c.localParticipant.Identity, "trackHandle", msg.DataTrackDemandUpdate.PubHandle, - "subscriberCount", msg.DataTrackDemandUpdate.SubscriberCount, + "hasSubscribers", msg.DataTrackDemandUpdate.HasSubscribers, ) c.lock.Lock() - c.dataTrackDemands[uint16(msg.DataTrackDemandUpdate.PubHandle)] = msg.DataTrackDemandUpdate.SubscriberCount + c.dataTrackDemands[uint16(msg.DataTrackDemandUpdate.PubHandle)] = msg.DataTrackDemandUpdate.HasSubscribers c.lock.Unlock() } return nil @@ -781,7 +781,7 @@ func (c *RTCClient) SubscribedDataTracks() map[livekit.ParticipantID]map[uint16] return tracks } -func (c *RTCClient) DataTrackDemands() map[uint16]uint32 { +func (c *RTCClient) DataTrackDemands() map[uint16]bool { c.lock.Lock() defer c.lock.Unlock() return maps.Clone(c.dataTrackDemands) @@ -1057,15 +1057,7 @@ func (c *RTCClient) PublishDataUnlabeled(data []byte) error { return c.publisher.SendDataMessageUnlabeled(data, true, "test") } -type PublishDataTrackOption func(*livekit.PublishDataTrackRequest) - -func PublishDataTrackDynacast() PublishDataTrackOption { - return func(req *livekit.PublishDataTrackRequest) { - req.IsDynacasted = true - } -} - -func (c *RTCClient) PublishDataTrack(opts ...PublishDataTrackOption) (writer TrackWriter, err error) { +func (c *RTCClient) PublishDataTrack() (writer TrackWriter, err error) { if err = c.ensurePublisherConnected(); err != nil { return } @@ -1075,9 +1067,6 @@ func (c *RTCClient) PublishDataTrack(opts ...PublishDataTrackOption) (writer Tra PubHandle: uint32(dataTrackHandle), Name: fmt.Sprintf("data_track_%d", dataTrackHandle), } - for _, opt := range opts { - opt(req) - } if err = c.SendRequest(&livekit.SignalRequest{ Message: &livekit.SignalRequest_PublishDataTrackRequest{ PublishDataTrackRequest: req, diff --git a/test/singlenode_test.go b/test/singlenode_test.go index e168c96cb..642e68ce9 100644 --- a/test/singlenode_test.go +++ b/test/singlenode_test.go @@ -1278,8 +1278,7 @@ func TestSinglePublisherDataTrack(t *testing.T) { require.NoError(t, err) defer dt1.Stop() - // publish the second track with dynacast enabled so c1 receives demand updates for it - dt2, err := c1.PublishDataTrack(testclient.PublishDataTrackDynacast()) + dt2, err := c1.PublishDataTrack() require.NoError(t, err) defer dt2.Stop() @@ -1294,18 +1293,18 @@ func TestSinglePublisherDataTrack(t *testing.T) { return "" }) - // c1 should get a demand update for the dynacasted track only + // c1 should get a demand update for both tracks once c2 subscribes testutils.WithTimeout(t, func() string { demands := c1.DataTrackDemands() - if len(demands) > 1 { - return "received demand update for non-dynacasted track" + if len(demands) != 2 { + return "did not receive demand updates for both tracks" } - for _, count := range demands { - if count == 1 { - return "" + for _, hasSubscribers := range demands { + if !hasSubscribers { + return "demand update did not indicate subscribers" } } - return "did not receive demand update for dynacasted track" + return "" }) // a new client joins and should get the initial stream @@ -1330,16 +1329,6 @@ func TestSinglePublisherDataTrack(t *testing.T) { require.True(t, strings.HasPrefix(string(tr.ID()), "DTR_"), "data track should begin with DTR") } - // demand for the dynacasted track should reflect both subscribers - testutils.WithTimeout(t, func() string { - for _, count := range c1.DataTrackDemands() { - if count == 2 { - return "" - } - } - return "demand did not reach 2 after c3 subscribed" - }) - // when c3 disconnects, ensure subscriber is cleaned up correctly c3.Stop() @@ -1356,14 +1345,15 @@ func TestSinglePublisherDataTrack(t *testing.T) { return "" }) - // demand for the dynacasted track should drop back to 1 + // once the last subscriber disconnects, demand should drop on both tracks + c2.Stop() testutils.WithTimeout(t, func() string { - for _, count := range c1.DataTrackDemands() { - if count == 1 { - return "" + for _, hasSubscribers := range c1.DataTrackDemands() { + if hasSubscribers { + return "demand did not drop after last subscriber disconnected" } } - return "demand did not drop back to 1 after c3 disconnected" + return "" }) }) }