diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index da9d46fef..3789afb29 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -3121,12 +3121,12 @@ func (p *ParticipantImpl) SendDataMessage(kind livekit.DataPacket_Kind, data []b return p.TransportManager.SendDataMessage(kind, data) } -func (p *ParticipantImpl) SendDataMessageUnlabeled(data []byte) error { +func (p *ParticipantImpl) SendDataMessageUnlabeled(data []byte, useRaw bool) error { if p.State() != livekit.ParticipantInfo_ACTIVE { return ErrDataChannelUnavailable } - return p.TransportManager.SendDataMessageUnlabeled(data) + return p.TransportManager.SendDataMessageUnlabeled(data, useRaw) } func (p *ParticipantImpl) onDataSendError(err error) { diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 9797a0b07..ebf093ec9 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -1842,7 +1842,7 @@ func BroadcastDataMessageForRoom(r types.Room, source types.LocalParticipant, da return } - op.SendDataMessageUnlabeled(data) + op.SendDataMessageUnlabeled(data, false) }) } diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 3c25fe182..6f1e502e6 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -35,6 +35,7 @@ import ( "github.com/pkg/errors" "go.uber.org/atomic" "go.uber.org/zap/zapcore" + "google.golang.org/protobuf/proto" "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/rtc/transport" @@ -1065,6 +1066,7 @@ func (t *PCTransport) CreateReadableDataChannel(label string, dci *webrtc.DataCh buffer := make([]byte, dataChannelBufferSize) for { n, _, err := rawDC.ReadDataChannel(buffer) + t.params.Logger.Infow("RAJA read data", "data", string(buffer[:n])) // REMVOE if err != nil { if !errors.Is(err, io.EOF) && !strings.Contains(err.Error(), "state=Closed") { t.params.Logger.Warnw("error reading data channel", err, "label", dc.Label()) @@ -1144,27 +1146,73 @@ func (t *PCTransport) WriteRTCP(pkts []rtcp.Packet) error { } func (t *PCTransport) SendDataMessage(kind livekit.DataPacket_Kind, data []byte) error { + convertFromUserPacket := false var dc *datachannel.DataChannelWriter[*webrtc.DataChannel] t.lock.RLock() - if kind == livekit.DataPacket_RELIABLE { - dc = t.reliableDC + if t.params.UseOneShotSignallingMode { + if len(t.unlabeledDataChannels) > 0 { + // use the first unlabeled to send + dc = t.unlabeledDataChannels[0] + } + convertFromUserPacket = true } else { - dc = t.lossyDC + if kind == livekit.DataPacket_RELIABLE { + dc = t.reliableDC + } else { + dc = t.lossyDC + } } t.lock.RUnlock() + if convertFromUserPacket { + dp := &livekit.DataPacket{} + if err := proto.Unmarshal(data, dp); err != nil { + return err + } + + switch payload := dp.Value.(type) { + case *livekit.DataPacket_User: + return t.sendDataMessage(dc, payload.User.Payload) + default: + return errors.New("cannot forward non user data packet") + } + } + return t.sendDataMessage(dc, data) } -func (t *PCTransport) SendDataMessageUnlabeled(data []byte) error { +func (t *PCTransport) SendDataMessageUnlabeled(data []byte, useRaw bool) error { + convertToUserPacket := false var dc *datachannel.DataChannelWriter[*webrtc.DataChannel] t.lock.RLock() - if len(t.unlabeledDataChannels) > 0 { - // use the first unlabeled to send - dc = t.unlabeledDataChannels[0] + 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 } t.lock.RUnlock() + if convertToUserPacket { + dpData, err := proto.Marshal(&livekit.DataPacket{ + 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 b6a0cc7ef..6475666ff 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -302,9 +302,13 @@ 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) error { +func (t *TransportManager) SendDataMessageUnlabeled(data []byte, useRaw bool) error { // downstream data is sent via primary peer connection - return t.handleSendDataResult(t.getTransport(true).SendDataMessageUnlabeled(data), "unlabeled", len(data)) + return t.handleSendDataResult( + t.getTransport(true).SendDataMessageUnlabeled(data, useRaw), + "unlabeled", + len(data), + ) } func (t *TransportManager) handleSendDataResult(err error, kind string, size int) error { diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 62ffc651e..31a10f9dd 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -407,7 +407,7 @@ type LocalParticipant interface { SendParticipantUpdate(participants []*livekit.ParticipantInfo) error SendSpeakerUpdate(speakers []*livekit.SpeakerInfo, force bool) error SendDataMessage(kind livekit.DataPacket_Kind, data []byte) error - SendDataMessageUnlabeled(data []byte) error + SendDataMessageUnlabeled(data []byte, useRaw bool) error SendRoomUpdate(room *livekit.Room) error SendConnectionQualityUpdate(update *livekit.ConnectionQualityUpdate) error SubscriptionPermissionUpdate(publisherID livekit.ParticipantID, trackID livekit.TrackID, allowed bool) diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index b0174fdbc..bb6eaa50b 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -843,10 +843,11 @@ type FakeLocalParticipant struct { sendDataMessageReturnsOnCall map[int]struct { result1 error } - SendDataMessageUnlabeledStub func([]byte) error + SendDataMessageUnlabeledStub func([]byte, bool) error sendDataMessageUnlabeledMutex sync.RWMutex sendDataMessageUnlabeledArgsForCall []struct { arg1 []byte + arg2 bool } sendDataMessageUnlabeledReturns struct { result1 error @@ -5665,7 +5666,7 @@ func (fake *FakeLocalParticipant) SendDataMessageReturnsOnCall(i int, result1 er }{result1} } -func (fake *FakeLocalParticipant) SendDataMessageUnlabeled(arg1 []byte) error { +func (fake *FakeLocalParticipant) SendDataMessageUnlabeled(arg1 []byte, arg2 bool) error { var arg1Copy []byte if arg1 != nil { arg1Copy = make([]byte, len(arg1)) @@ -5675,13 +5676,14 @@ func (fake *FakeLocalParticipant) SendDataMessageUnlabeled(arg1 []byte) error { ret, specificReturn := fake.sendDataMessageUnlabeledReturnsOnCall[len(fake.sendDataMessageUnlabeledArgsForCall)] fake.sendDataMessageUnlabeledArgsForCall = append(fake.sendDataMessageUnlabeledArgsForCall, struct { arg1 []byte - }{arg1Copy}) + arg2 bool + }{arg1Copy, arg2}) stub := fake.SendDataMessageUnlabeledStub fakeReturns := fake.sendDataMessageUnlabeledReturns - fake.recordInvocation("SendDataMessageUnlabeled", []interface{}{arg1Copy}) + fake.recordInvocation("SendDataMessageUnlabeled", []interface{}{arg1Copy, arg2}) fake.sendDataMessageUnlabeledMutex.Unlock() if stub != nil { - return stub(arg1) + return stub(arg1, arg2) } if specificReturn { return ret.result1 @@ -5695,17 +5697,17 @@ func (fake *FakeLocalParticipant) SendDataMessageUnlabeledCallCount() int { return len(fake.sendDataMessageUnlabeledArgsForCall) } -func (fake *FakeLocalParticipant) SendDataMessageUnlabeledCalls(stub func([]byte) error) { +func (fake *FakeLocalParticipant) SendDataMessageUnlabeledCalls(stub func([]byte, bool) error) { fake.sendDataMessageUnlabeledMutex.Lock() defer fake.sendDataMessageUnlabeledMutex.Unlock() fake.SendDataMessageUnlabeledStub = stub } -func (fake *FakeLocalParticipant) SendDataMessageUnlabeledArgsForCall(i int) []byte { +func (fake *FakeLocalParticipant) SendDataMessageUnlabeledArgsForCall(i int) ([]byte, bool) { fake.sendDataMessageUnlabeledMutex.RLock() defer fake.sendDataMessageUnlabeledMutex.RUnlock() argsForCall := fake.sendDataMessageUnlabeledArgsForCall[i] - return argsForCall.arg1 + return argsForCall.arg1, argsForCall.arg2 } func (fake *FakeLocalParticipant) SendDataMessageUnlabeledReturns(result1 error) { diff --git a/test/client/client.go b/test/client/client.go index 3b2d17464..3e51ae4d2 100644 --- a/test/client/client.go +++ b/test/client/client.go @@ -754,8 +754,7 @@ func (c *RTCClient) PublishDataUnlabeled(data []byte) error { return err } - fmt.Printf("RAJA sending unlabeled data: %s\n", string(data)) // REMOVE - return c.publisher.SendDataMessageUnlabeled(data) + return c.publisher.SendDataMessageUnlabeled(data, true) } func (c *RTCClient) GetPublishedTrackIDs() []string { diff --git a/test/scenarios.go b/test/scenarios.go index d1aa8f0b3..9a00776fd 100644 --- a/test/scenarios.go +++ b/test/scenarios.go @@ -173,8 +173,7 @@ func scenarioDataUnlabeledPublish(t *testing.T) { payload := "test unlabeled bytes" received := atomic.NewBool(false) - c2.OnDataUnlabeledReceived = func(data []byte) { - fmt.Printf("RAJA received data: message: %s, sid: %s\n", string(data), c1.ID()) // REMOVE + c2.OnDataReceived = func(data []byte, _sid string) { if string(data) == payload { received.Store(true) }