Update tests

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,

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 ""
 			})
 		})
 	}