diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 690db92be..aaddcd81d 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -4278,11 +4278,24 @@ func (p *ParticipantImpl) SendDataMessage(kind livekit.DataPacket_Kind, data []b } func (p *ParticipantImpl) SendDataMessageUnlabeled(data []byte, useRaw bool, sender livekit.ParticipantIdentity) error { - if p.State() != livekit.ParticipantInfo_ACTIVE { - return ErrDataChannelUnavailable + if useRaw || p.params.UseOneShotSignallingMode { + if p.State() != livekit.ParticipantInfo_ACTIVE { + return ErrDataChannelUnavailable + } + return p.TransportManager.SendDataMessageUnlabeled(data) } - return p.TransportManager.SendDataMessageUnlabeled(data, useRaw, sender) + // a UserPacket on the reliable channel, held like any reliable message while that channel opens + dpData, err := proto.Marshal(&livekit.DataPacket{ + ParticipantIdentity: string(sender), + Value: &livekit.DataPacket_User{ + User: &livekit.UserPacket{Payload: data}, + }, + }) + if err != nil { + return err + } + return p.SendDataMessage(livekit.DataPacket_RELIABLE, dpData, "", 0) } func (p *ParticipantImpl) onDataSendError(err error) { diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index 4dfe5148c..04f296033 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -1325,6 +1325,18 @@ func TestUnsequencedReliableDataBufferedWhileJoining(t *testing.T) { require.Error(t, p.SendDataMessage(livekit.DataPacket_LOSSY, []byte("one"), "", 0)) require.Empty(t, p.reliableDataInfo.joiningUnsequencedMessages) }) + + t.Run("unlabeled data is buffered as a user packet", func(t *testing.T) { + p := newParticipantForTest("test") + + require.NoError(t, p.SendDataMessageUnlabeled([]byte("one"), false, "alice")) + + require.Len(t, p.reliableDataInfo.joiningUnsequencedMessages, 1) + dp := &livekit.DataPacket{} + require.NoError(t, proto.Unmarshal(p.reliableDataInfo.joiningUnsequencedMessages[0], dp)) + require.Equal(t, "alice", dp.ParticipantIdentity) + require.Equal(t, []byte("one"), dp.GetUser().GetPayload()) + }) } func TestMigratingInParticipantWaitsForReconnectResponse(t *testing.T) { diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 1632ea508..d72a04ecb 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -1537,39 +1537,15 @@ func (t *PCTransport) SendDataMessage(kind livekit.DataPacket_Kind, data []byte) return t.sendDataMessage(dc, data) } -func (t *PCTransport) SendDataMessageUnlabeled(data []byte, useRaw bool, sender livekit.ParticipantIdentity) error { - convertToUserPacket := false +func (t *PCTransport) SendDataMessageUnlabeled(data []byte) error { var dc *datachannel.DataChannelWriter[*webrtc.DataChannel] t.lock.RLock() - if t.params.UseOneShotSignallingMode || useRaw { - if len(t.unlabeledDataChannels) > 0 { - // use the first unlabeled to send - dc = t.unlabeledDataChannels[0] - } - } else { - if t.reliableDC != nil { - dc = t.reliableDC - } else if t.lossyDC != nil { - dc = t.lossyDC - } - - convertToUserPacket = true + if len(t.unlabeledDataChannels) > 0 { + // use the first unlabeled to send + dc = t.unlabeledDataChannels[0] } t.lock.RUnlock() - if convertToUserPacket { - dpData, err := proto.Marshal(&livekit.DataPacket{ - ParticipantIdentity: string(sender), - Value: &livekit.DataPacket_User{ - User: &livekit.UserPacket{Payload: data}, - }, - }) - if err != nil { - return err - } - return t.sendDataMessage(dc, dpData) - } - return t.sendDataMessage(dc, data) } diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index fad77024f..2aac96fe2 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -376,13 +376,9 @@ func (t *TransportManager) SendDataMessage(kind livekit.DataPacket_Kind, data [] return t.handleSendDataResult(t.getTransport(true).SendDataMessage(kind, data), kind.String(), len(data)) } -func (t *TransportManager) SendDataMessageUnlabeled(data []byte, useRaw bool, sender livekit.ParticipantIdentity) error { +func (t *TransportManager) SendDataMessageUnlabeled(data []byte) error { // downstream data is sent via primary peer connection - return t.handleSendDataResult( - t.getTransport(true).SendDataMessageUnlabeled(data, useRaw, sender), - "unlabeled", - len(data), - ) + return t.handleSendDataResult(t.getTransport(true).SendDataMessageUnlabeled(data), "unlabeled", len(data)) } func (t *TransportManager) handleSendDataResult(err error, kind string, size int) error { diff --git a/test/client/client.go b/test/client/client.go index 66030dc4a..9f1e4a65c 100644 --- a/test/client/client.go +++ b/test/client/client.go @@ -1066,7 +1066,7 @@ func (c *RTCClient) PublishDataUnlabeled(data []byte) error { return err } - return c.publisher.SendDataMessageUnlabeled(data, true, "test") + return c.publisher.SendDataMessageUnlabeled(data) } func (c *RTCClient) PublishDataTrack() (writer TrackWriter, err error) {