diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 39f48560e..32bb0140f 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -2188,8 +2188,20 @@ func BroadcastDataMessageForRoom(r types.Room, source types.LocalParticipant, da sourceIdentity = source.Identity() } + BroadcastDataMessageForRoomFrom(r, sourceID, sourceIdentity, data, logger) +} + +// BroadcastDataMessageForRoomFrom sends an unlabeled data message to every local participant, stamping +// sourceIdentity as the sender and skipping the local participant with sourceID, if there is one. +func BroadcastDataMessageForRoomFrom( + r types.Room, + sourceID livekit.ParticipantID, + sourceIdentity livekit.ParticipantIdentity, + data []byte, + logger logger.Logger, +) { utils.ParallelExec(r.GetLocalParticipants(), dataForwardLoadBalanceThreshold, 1, func(op types.LocalParticipant) { - if source != nil && op.ID() == sourceID { + if sourceID != "" && op.ID() == sourceID { return } diff --git a/pkg/rtc/room_test.go b/pkg/rtc/room_test.go index d844733c6..d4e2db3b4 100644 --- a/pkg/rtc/room_test.go +++ b/pkg/rtc/room_test.go @@ -788,6 +788,38 @@ func TestDataChannel(t *testing.T) { } }) + t.Run("unlabeled broadcast from a remote identity", func(t *testing.T) { + rm := newRoomWithParticipants(t, testRoomOpts{num: 2}) + defer rm.Close(types.RoomCloseReasonUnknown) + + BroadcastDataMessageForRoomFrom(rm, "", "remote", []byte("relayed"), rm.logger) + + for _, op := range rm.GetParticipants() { + fp := op.(*typesfakes.FakeLocalParticipant) + require.Equal(t, 1, fp.SendDataMessageUnlabeledCallCount()) + got, _, sender := fp.SendDataMessageUnlabeledArgsForCall(0) + require.Equal(t, []byte("relayed"), got) + require.Equal(t, livekit.ParticipantIdentity("remote"), sender) + } + }) + + t.Run("unlabeled broadcast from a local id skips the source", func(t *testing.T) { + rm := newRoomWithParticipants(t, testRoomOpts{num: 2}) + defer rm.Close(types.RoomCloseReasonUnknown) + + participants := rm.GetParticipants() + p := participants[0].(*typesfakes.FakeLocalParticipant) + BroadcastDataMessageForRoomFrom(rm, p.ID(), p.Identity(), []byte("local"), rm.logger) + + require.Zero(t, p.SendDataMessageUnlabeledCallCount()) + for _, op := range participants[1:] { + fp := op.(*typesfakes.FakeLocalParticipant) + require.Equal(t, 1, fp.SendDataMessageUnlabeledCallCount()) + _, _, sender := fp.SendDataMessageUnlabeledArgsForCall(0) + require.Equal(t, p.Identity(), sender) + } + }) + t.Run("publishing disallowed", func(t *testing.T) { rm := newRoomWithParticipants(t, testRoomOpts{num: 2}) defer rm.Close(types.RoomCloseReasonUnknown)