dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-10-15 09:00:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

Forward data between WHIP client and non-WHIP client (#3627)

Browse Source 
* Forward data between WHIP client and non-WHIP client

* fix test
This commit is contained in:
Raja Subramanian
2025-04-25 23:20:08 +05:30
committed by GitHub
parent 6739e7bcc8
commit e14905581c
8 changed files with 77 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/participant.go
+2 -2
View File
@@ -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) {
pkg/rtc/room.go
+1 -1
View File
@@ -1842,7 +1842,7 @@ func BroadcastDataMessageForRoom(r types.Room, source types.LocalParticipant, da
return
}
op.SendDataMessageUnlabeled(data)
op.SendDataMessageUnlabeled(data, false)
})
}
pkg/rtc/transport.go
+55 -7
View File
@@ -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)
}
pkg/rtc/transportmanager.go
+6 -2
View File
@@ -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 {
pkg/rtc/types/interfaces.go
+1 -1
View File
@@ -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)
pkg/rtc/types/typesfakes/fake_local_participant.go
+10 -8
View File
@@ -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) {
test/client/client.go
+1 -2
View File
@@ -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 {
test/scenarios.go
+1 -2
View File
@@ -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)
}