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

get full definitions, not just ids

Browse Source
This commit is contained in:
boks1971
2026-05-24 23:21:13 +05:30
parent 038f086a78
commit 350d5cc280
7 changed files with 50 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/participant_async_attributes.go
-12
View File
@@ -91,18 +91,6 @@ func (p *ParticipantAsyncAttributes) GetAll() []*livekit.DataTrackSchemaDefiniti
return all
}
func (p *ParticipantAsyncAttributes) GetAllIDs() []*livekit.DataTrackSchemaId {
p.lock.Lock()
defer p.lock.Unlock()
ids := make([]*livekit.DataTrackSchemaId, 0, len(p.attributes))
for _, aa := range p.attributes {
ids = append(ids, utils.CloneProto(aa.Id))
}
return ids
}
// -------------------------------
func ToParticipantAsyncAttributeKey(id *livekit.DataTrackSchemaId) string {
pkg/rtc/participant_async_attributes_handler.go
+2 -2
View File
@@ -121,6 +121,6 @@ func (p *ParticipantImpl) ProcessGetDataTrackSchemaRequest(req *livekit.GetDataT
p.sendGetDataTrackSchemaResponse(asyncAttribute)
}
func (p *ParticipantImpl) GetAllAsyncAttributeIDs() []*livekit.DataTrackSchemaId {
return p.asyncAttributes.GetAllIDs()
func (p *ParticipantImpl) GetAllAsyncAttributes() []*livekit.DataTrackSchemaDefinition {
return p.asyncAttributes.GetAll()
}
pkg/rtc/participant_async_attributes_handler_test.go
-72
View File
@@ -321,75 +321,3 @@ func TestProcessGetDataTrackSchemaRequest(t *testing.T) {
require.Equal(t, def, response.GetDataTrackSchemaResponse.SchemaDefinition)
})
}
func TestGetAllAsyncAttributeIDs(t *testing.T) {
t.Run("returns empty slice when no schemas defined", func(t *testing.T) {
p := newParticipantWithAsyncAttributes(t, true, 0)
require.Empty(t, p.GetAllAsyncAttributeIDs())
})
t.Run("returns ids for all defined schemas", func(t *testing.T) {
p := newParticipantWithAsyncAttributes(t, true, 0)
id1 := &livekit.DataTrackSchemaId{
Name: "schema-1",
Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF,
}
id2 := &livekit.DataTrackSchemaId{
Name: "schema-2",
Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_FLATBUFFER,
}
id3 := &livekit.DataTrackSchemaId{
Name: "schema-1",
Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_JSON_SCHEMA,
}
p.AddDataTrackSchema(&livekit.DataTrackSchemaDefinition{Id: id1, Definition: []byte("def-1")})
p.AddDataTrackSchema(&livekit.DataTrackSchemaDefinition{Id: id2, Definition: []byte("def-2")})
p.AddDataTrackSchema(&livekit.DataTrackSchemaDefinition{Id: id3, Definition: []byte("def-3")})
ids := p.GetAllAsyncAttributeIDs()
require.Len(t, ids, 3)
got := make(map[string]livekit.DataTrackSchemaEncoding, len(ids))
for _, id := range ids {
got[id.Name+"|"+id.Encoding.String()] = id.Encoding
}
require.Contains(t, got, "schema-1|"+id1.Encoding.String())
require.Contains(t, got, "schema-2|"+id2.Encoding.String())
require.Contains(t, got, "schema-1|"+id3.Encoding.String())
})
t.Run("reflects deletes", func(t *testing.T) {
p := newParticipantWithAsyncAttributes(t, true, 0)
id := &livekit.DataTrackSchemaId{
Name: "schema-1",
Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF,
}
p.AddDataTrackSchema(&livekit.DataTrackSchemaDefinition{Id: id, Definition: []byte("def")})
require.Len(t, p.GetAllAsyncAttributeIDs(), 1)
p.asyncAttributes.Delete(id)
require.Empty(t, p.GetAllAsyncAttributeIDs())
})
t.Run("returned ids are cloned and do not alias internal state", func(t *testing.T) {
p := newParticipantWithAsyncAttributes(t, true, 0)
id := &livekit.DataTrackSchemaId{
Name: "schema-1",
Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF,
}
p.AddDataTrackSchema(&livekit.DataTrackSchemaDefinition{Id: id, Definition: []byte("def")})
ids := p.GetAllAsyncAttributeIDs()
require.Len(t, ids, 1)
ids[0].Name = "mutated"
// mutating the returned slice must not affect what's stored
again := p.GetAllAsyncAttributeIDs()
require.Len(t, again, 1)
require.Equal(t, "schema-1", again[0].Name)
})
}
pkg/rtc/participant_async_attributes_test.go
-5
View File
@@ -63,10 +63,6 @@ func TestParticipantAsyncAttributes_AddOverwrites(t *testing.T) {
require.Equal(t, []byte("v2"), got.Definition)
require.Len(t, a.GetAll(), 1)
allIDs := a.GetAllIDs()
require.Len(t, allIDs, 1)
require.Equal(t, id.Name, allIDs[0].Name)
require.Equal(t, id.Encoding, allIDs[0].Encoding)
}
func TestParticipantAsyncAttributes_DifferentEncodingsAreDistinct(t *testing.T) {
@@ -93,7 +89,6 @@ func TestParticipantAsyncAttributes_DifferentEncodingsAreDistinct(t *testing.T)
require.Equal(t, []byte("json-def"), gotJSON.Definition)
require.Len(t, a.GetAll(), 2)
require.Len(t, a.GetAllIDs(), 2)
}
func TestParticipantAsyncAttributes_Delete(t *testing.T) {
pkg/rtc/types/interfaces.go
+1 -1
View File
@@ -577,7 +577,7 @@ type LocalParticipant interface {
GetNextSubscribedDataTrackHandle() uint16
GetAllAsyncAttributeIDs() []*livekit.DataTrackSchemaId
GetAllAsyncAttributes() []*livekit.DataTrackSchemaDefinition
}
// ---------------------------------------------
pkg/rtc/types/typesfakes/fake_local_participant.go
+38 -38
View File
@@ -221,15 +221,15 @@ type FakeLocalParticipant struct {
getAdaptiveStreamReturnsOnCall map[int]struct {
result1 bool
}
GetAllAsyncAttributeIDsStub func() []*livekit.DataTrackSchemaId
getAllAsyncAttributeIDsMutex sync.RWMutex
getAllAsyncAttributeIDsArgsForCall []struct {
GetAllAsyncAttributesStub func() []*livekit.DataTrackSchemaDefinition
getAllAsyncAttributesMutex sync.RWMutex
getAllAsyncAttributesArgsForCall []struct {
}
getAllAsyncAttributeIDsReturns struct {
result1 []*livekit.DataTrackSchemaId
getAllAsyncAttributesReturns struct {
result1 []*livekit.DataTrackSchemaDefinition
}
getAllAsyncAttributeIDsReturnsOnCall map[int]struct {
result1 []*livekit.DataTrackSchemaId
getAllAsyncAttributesReturnsOnCall map[int]struct {
result1 []*livekit.DataTrackSchemaDefinition
}
GetAnswerStub func() (webrtc.SessionDescription, uint32, error)
getAnswerMutex sync.RWMutex
@@ -2593,15 +2593,15 @@ func (fake *FakeLocalParticipant) GetAdaptiveStreamReturnsOnCall(i int, result1
}{result1}
}
func (fake *FakeLocalParticipant) GetAllAsyncAttributeIDs() []*livekit.DataTrackSchemaId {
fake.getAllAsyncAttributeIDsMutex.Lock()
ret, specificReturn := fake.getAllAsyncAttributeIDsReturnsOnCall[len(fake.getAllAsyncAttributeIDsArgsForCall)]
fake.getAllAsyncAttributeIDsArgsForCall = append(fake.getAllAsyncAttributeIDsArgsForCall, struct {
func (fake *FakeLocalParticipant) GetAllAsyncAttributes() []*livekit.DataTrackSchemaDefinition {
fake.getAllAsyncAttributesMutex.Lock()
ret, specificReturn := fake.getAllAsyncAttributesReturnsOnCall[len(fake.getAllAsyncAttributesArgsForCall)]
fake.getAllAsyncAttributesArgsForCall = append(fake.getAllAsyncAttributesArgsForCall, struct {
}{})
stub := fake.GetAllAsyncAttributeIDsStub
fakeReturns := fake.getAllAsyncAttributeIDsReturns
fake.recordInvocation("GetAllAsyncAttributeIDs", []interface{}{})
fake.getAllAsyncAttributeIDsMutex.Unlock()
stub := fake.GetAllAsyncAttributesStub
fakeReturns := fake.getAllAsyncAttributesReturns
fake.recordInvocation("GetAllAsyncAttributes", []interface{}{})
fake.getAllAsyncAttributesMutex.Unlock()
if stub != nil {
return stub()
}
@@ -2611,38 +2611,38 @@ func (fake *FakeLocalParticipant) GetAllAsyncAttributeIDs() []*livekit.DataTrack
return fakeReturns.result1
}
func (fake *FakeLocalParticipant) GetAllAsyncAttributeIDsCallCount() int {
fake.getAllAsyncAttributeIDsMutex.RLock()
defer fake.getAllAsyncAttributeIDsMutex.RUnlock()
return len(fake.getAllAsyncAttributeIDsArgsForCall)
func (fake *FakeLocalParticipant) GetAllAsyncAttributesCallCount() int {
fake.getAllAsyncAttributesMutex.RLock()
defer fake.getAllAsyncAttributesMutex.RUnlock()
return len(fake.getAllAsyncAttributesArgsForCall)
}
func (fake *FakeLocalParticipant) GetAllAsyncAttributeIDsCalls(stub func() []*livekit.DataTrackSchemaId) {
fake.getAllAsyncAttributeIDsMutex.Lock()
defer fake.getAllAsyncAttributeIDsMutex.Unlock()
fake.GetAllAsyncAttributeIDsStub = stub
func (fake *FakeLocalParticipant) GetAllAsyncAttributesCalls(stub func() []*livekit.DataTrackSchemaDefinition) {
fake.getAllAsyncAttributesMutex.Lock()
defer fake.getAllAsyncAttributesMutex.Unlock()
fake.GetAllAsyncAttributesStub = stub
}
func (fake *FakeLocalParticipant) GetAllAsyncAttributeIDsReturns(result1 []*livekit.DataTrackSchemaId) {
fake.getAllAsyncAttributeIDsMutex.Lock()
defer fake.getAllAsyncAttributeIDsMutex.Unlock()
fake.GetAllAsyncAttributeIDsStub = nil
fake.getAllAsyncAttributeIDsReturns = struct {
result1 []*livekit.DataTrackSchemaId
func (fake *FakeLocalParticipant) GetAllAsyncAttributesReturns(result1 []*livekit.DataTrackSchemaDefinition) {
fake.getAllAsyncAttributesMutex.Lock()
defer fake.getAllAsyncAttributesMutex.Unlock()
fake.GetAllAsyncAttributesStub = nil
fake.getAllAsyncAttributesReturns = struct {
result1 []*livekit.DataTrackSchemaDefinition
}{result1}
}
func (fake *FakeLocalParticipant) GetAllAsyncAttributeIDsReturnsOnCall(i int, result1 []*livekit.DataTrackSchemaId) {
fake.getAllAsyncAttributeIDsMutex.Lock()
defer fake.getAllAsyncAttributeIDsMutex.Unlock()
fake.GetAllAsyncAttributeIDsStub = nil
if fake.getAllAsyncAttributeIDsReturnsOnCall == nil {
fake.getAllAsyncAttributeIDsReturnsOnCall = make(map[int]struct {
result1 []*livekit.DataTrackSchemaId
func (fake *FakeLocalParticipant) GetAllAsyncAttributesReturnsOnCall(i int, result1 []*livekit.DataTrackSchemaDefinition) {
fake.getAllAsyncAttributesMutex.Lock()
defer fake.getAllAsyncAttributesMutex.Unlock()
fake.GetAllAsyncAttributesStub = nil
if fake.getAllAsyncAttributesReturnsOnCall == nil {
fake.getAllAsyncAttributesReturnsOnCall = make(map[int]struct {
result1 []*livekit.DataTrackSchemaDefinition
})
}
fake.getAllAsyncAttributeIDsReturnsOnCall[i] = struct {
result1 []*livekit.DataTrackSchemaId
fake.getAllAsyncAttributesReturnsOnCall[i] = struct {
result1 []*livekit.DataTrackSchemaDefinition
}{result1}
}
pkg/service/wire_gen.go
+9 -9
View File
@@ -90,23 +90,23 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
}
rtcEgressLauncher := NewEgressLauncher(egressClient, ioInfoService, objectStore)
topicFormatter := rpc.NewTopicFormatter()
roomClient, err := rpc.NewTypedRoomClient(clientParams)
v, err := rpc.NewTypedRoomClient(clientParams)
if err != nil {
return nil, err
}
participantClient, err := rpc.NewTypedParticipantClient(clientParams)
v2, err := rpc.NewTypedParticipantClient(clientParams)
if err != nil {
return nil, err
}
roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, rtcEgressLauncher, topicFormatter, roomClient, participantClient)
roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, rtcEgressLauncher, topicFormatter, v, v2)
if err != nil {
return nil, err
}
agentDispatchInternalClient, err := rpc.NewTypedAgentDispatchInternalClient(clientParams)
v3, err := rpc.NewTypedAgentDispatchInternalClient(clientParams)
if err != nil {
return nil, err
}
agentDispatchService := NewAgentDispatchService(agentDispatchInternalClient, topicFormatter, roomAllocator, router)
agentDispatchService := NewAgentDispatchService(v3, topicFormatter, roomAllocator, router)
egressService := NewEgressService(egressClient, rtcEgressLauncher, ioInfoService, roomService)
ingressConfig := getIngressConfig(conf)
ingressClient, err := rpc.NewIngressClient(clientParams)
@@ -121,11 +121,11 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
}
sipService := NewSIPService(sipConfig, nodeID, messageBus, sipClient, sipStore, roomService, telemetryService)
rtcService := NewRTCService(conf, roomAllocator, router, telemetryService)
whipParticipantClient, err := rpc.NewTypedWHIPParticipantClient(clientParams)
v4, err := rpc.NewTypedWHIPParticipantClient(clientParams)
if err != nil {
return nil, err
}
serviceWHIPService, err := NewWHIPService(conf, router, roomAllocator, clientParams, topicFormatter, whipParticipantClient)
serviceWHIPService, err := NewWHIPService(conf, router, roomAllocator, clientParams, topicFormatter, v4)
if err != nil {
return nil, err
}
@@ -150,8 +150,8 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
if err != nil {
return nil, err
}
authHandler := getTURNAuthHandlerFunc(turnAuthHandler)
server, err := newInProcessTurnServer(conf, authHandler)
v5 := getTURNAuthHandlerFunc(turnAuthHandler)
server, err := newInProcessTurnServer(conf, v5)
if err != nil {
return nil, err
}