diff --git a/pkg/rtc/participant_async_attributes.go b/pkg/rtc/participant_async_attributes.go index b17bba526..46e7bdfe9 100644 --- a/pkg/rtc/participant_async_attributes.go +++ b/pkg/rtc/participant_async_attributes.go @@ -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 { diff --git a/pkg/rtc/participant_async_attributes_handler.go b/pkg/rtc/participant_async_attributes_handler.go index f5029f61c..63a156953 100644 --- a/pkg/rtc/participant_async_attributes_handler.go +++ b/pkg/rtc/participant_async_attributes_handler.go @@ -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() } diff --git a/pkg/rtc/participant_async_attributes_handler_test.go b/pkg/rtc/participant_async_attributes_handler_test.go index 97761e41f..c6c98be61 100644 --- a/pkg/rtc/participant_async_attributes_handler_test.go +++ b/pkg/rtc/participant_async_attributes_handler_test.go @@ -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) - }) -} diff --git a/pkg/rtc/participant_async_attributes_test.go b/pkg/rtc/participant_async_attributes_test.go index e64045108..5f68110d0 100644 --- a/pkg/rtc/participant_async_attributes_test.go +++ b/pkg/rtc/participant_async_attributes_test.go @@ -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) { diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index d6baaed88..7e67846fb 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -577,7 +577,7 @@ type LocalParticipant interface { GetNextSubscribedDataTrackHandle() uint16 - GetAllAsyncAttributeIDs() []*livekit.DataTrackSchemaId + GetAllAsyncAttributes() []*livekit.DataTrackSchemaDefinition } // --------------------------------------------- diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index d9c7e0a5b..dadab8550 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -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} } diff --git a/pkg/service/wire_gen.go b/pkg/service/wire_gen.go index 57b831a00..1bfe93000 100644 --- a/pkg/service/wire_gen.go +++ b/pkg/service/wire_gen.go @@ -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 }