From 422dbecbbbe27837c82dec08821339b8e0cd8e1f Mon Sep 17 00:00:00 2001 From: boks1971 Date: Sun, 24 May 2026 16:26:02 +0530 Subject: [PATCH] Async attributes on participant. How it is different from existing participant attributes? 1. Async attribute can be added one at a time. 2. These are not included in `ParticipantInfo`. 3. Get an attribute bt participant identity and async attribute ID as and when needed. --- go.mod | 2 + go.sum | 2 - pkg/config/config.go | 33 ++ pkg/rtc/participant.go | 106 +++--- pkg/rtc/participant_async_attributes.go | 105 ++++++ .../participant_async_attributes_handler.go | 126 +++++++ ...rticipant_async_attributes_handler_test.go | 319 ++++++++++++++++++ pkg/rtc/participant_async_attributes_test.go | 207 ++++++++++++ pkg/rtc/participant_signal.go | 6 + pkg/rtc/room.go | 12 + pkg/rtc/signalling/interfaces.go | 1 + pkg/rtc/signalling/signalhandler.go | 6 + pkg/rtc/signalling/signalling.go | 8 + pkg/rtc/signalling/signallingunimplemented.go | 4 + pkg/rtc/types/interfaces.go | 12 + .../typesfakes/fake_local_participant.go | 222 ++++++++++++ .../fake_local_participant_listener.go | 78 +++++ pkg/rtc/types/typesfakes/fake_participant.go | 109 ++++++ pkg/service/roommanager.go | 5 +- test/integration_helpers.go | 13 +- test/multinode_test.go | 111 ++++++ test/singlenode_test.go | 266 +++++++++++++++ 22 files changed, 1696 insertions(+), 57 deletions(-) create mode 100644 pkg/rtc/participant_async_attributes.go create mode 100644 pkg/rtc/participant_async_attributes_handler.go create mode 100644 pkg/rtc/participant_async_attributes_handler_test.go create mode 100644 pkg/rtc/participant_async_attributes_test.go diff --git a/go.mod b/go.mod index acba029a4..a63705061 100644 --- a/go.mod +++ b/go.mod @@ -164,3 +164,5 @@ require ( google.golang.org/grpc v1.80.0 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect ) + +replace github.com/livekit/protocol => ../protocol diff --git a/go.sum b/go.sum index 41cfe1273..82f637907 100644 --- a/go.sum +++ b/go.sum @@ -181,8 +181,6 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20260521171458-ef117e280447 h1:AwzxgDnzGVVLZuSYDdgo7ETdpHzQPagyjtJOTrhQduY= github.com/livekit/mediatransportutil v0.0.0-20260521171458-ef117e280447/go.mod h1:RCd46PT+6sEztld6XpkCrG1xskb0u3SqxIjy4G897Ss= -github.com/livekit/protocol v1.45.9-0.20260519061926-8381f2180c45 h1:gJQFJNjHuxeKroI6KTtVeXVcUMOaK8ksdiB6FoiDWmE= -github.com/livekit/protocol v1.45.9-0.20260519061926-8381f2180c45/go.mod h1:KEPIJ/ZdMFQ9tmmfv/uT9TjQEuEcZupCZBabuRGEC1k= github.com/livekit/psrpc v0.7.1 h1:ms37az0QTD3UXIWuUC5D/SkmKOlRMVRsI261eBWu/Vw= github.com/livekit/psrpc v0.7.1/go.mod h1:bZ4iHFQptTkbPnB0LasvRNu/OBYXEu1NA6O5BMFo9kk= github.com/mackerelio/go-osstat v0.2.7 h1:TCavZi10wF49bT6iQZ9eT2keGZQpC69MTDfdJej5e94= diff --git a/pkg/config/config.go b/pkg/config/config.go index f155eb59d..79219dae8 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -89,6 +89,8 @@ type Config struct { EnableDataTracks bool `yaml:"enable_data_tracks,omitempty"` + EnableParticipantAsyncAttributes bool `yaml:"enable_participant_async_attributes,omitempty"` + API APIConfig `yaml:"api,omitempty"` } @@ -265,6 +267,8 @@ type RegionConfig struct { Lon float64 `yaml:"lon,omitempty"` } +// --------------------------------- + type LimitConfig struct { NumTracks int32 `yaml:"num_tracks,omitempty"` BytesPerSec float32 `yaml:"bytes_per_sec,omitempty"` @@ -276,6 +280,8 @@ type LimitConfig struct { MaxRoomNameLength int `yaml:"max_room_name_length,omitempty"` MaxParticipantIdentityLength int `yaml:"max_participant_identity_length,omitempty"` MaxParticipantNameLength int `yaml:"max_participant_name_length,omitempty"` + + MaxAsyncAttributesSize uint32 `yaml:"max_async_attributes_size,omitempty"` } func (l LimitConfig) CheckRoomNameLength(name string) bool { @@ -306,6 +312,32 @@ func (l LimitConfig) CheckAttributesSize(attributes map[string]string) bool { return uint32(total) <= l.MaxAttributesSize } +func (l LimitConfig) CheckAsyncAttributesSize(asyncAttributes map[string][]byte) bool { + if l.MaxAsyncAttributesSize == 0 { + return true + } + + total := 0 + for k, v := range asyncAttributes { + total += len(k) + len(v) + } + return uint32(total) <= l.MaxAsyncAttributesSize +} + +func (l LimitConfig) CanAddAsyncAttribute(asyncAttributes map[string][]byte, toAddKey string, toAddValue []byte) bool { + if l.MaxAsyncAttributesSize == 0 { + return true + } + + total := 0 + for k, v := range asyncAttributes { + total += len(k) + len(v) + } + return uint32(total+len(toAddKey)+len(toAddValue)) <= l.MaxAsyncAttributesSize +} + +// --------------------------------- + type IngressConfig struct { RTMPBaseURL string `yaml:"rtmp_base_url,omitempty"` WHIPBaseURL string `yaml:"whip_base_url,omitempty"` @@ -425,6 +457,7 @@ var DefaultConfig = Config{ MaxRoomNameLength: 256, MaxParticipantIdentityLength: 256, MaxParticipantNameLength: 256, + MaxAsyncAttributesSize: 256000, }, Logging: LoggingConfig{ PionLevel: "error", diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 3c409f92a..a7548c8f0 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -174,56 +174,57 @@ type ParticipantParams struct { PLIThrottleConfig sfu.PLIThrottleConfig CongestionControlConfig config.CongestionControlConfig // codecs that are enabled for this room - PublishEnabledCodecs []*livekit.Codec - SubscribeEnabledCodecs []*livekit.Codec - Logger logger.Logger - LoggerResolver logger.DeferredFieldResolver - Reporter roomobs.ParticipantSessionReporter - ReporterResolver roomobs.ParticipantReporterResolver - SimTracks map[uint32]interceptor.SimulcastTrackInfo - Grants *auth.ClaimGrants - InitialVersion uint32 - ClientConf *livekit.ClientConfiguration - ClientInfo ClientInfo - Region string - Migration bool - Reconnect bool - AdaptiveStream bool - AllowTCPFallback bool - TCPFallbackRTTThreshold int - AllowUDPUnstableFallback bool - TURNSEnabled bool - ParticipantListener types.LocalParticipantListener - ParticipantHelper types.LocalParticipantHelper - DisableSupervisor bool - ReconnectOnPublicationError bool - ReconnectOnSubscriptionError bool - ReconnectOnDataChannelError bool - VersionGenerator utils.TimedVersionGenerator - DisableDynacast bool - SubscriberAllowPause bool - SubscriptionLimitAudio int32 - SubscriptionLimitVideo int32 - PlayoutDelay *livekit.PlayoutDelay - SyncStreams bool - ForwardStats *sfu.ForwardStats - DisableSenderReportPassThrough bool - MetricConfig metric.MetricConfig - UseOneShotSignallingMode bool - EnableMetrics bool - DataChannelMaxBufferedAmount uint64 - DatachannelSlowThreshold int - DatachannelLossyTargetLatency time.Duration - FireOnTrackBySdp bool - DisableCodecRegression bool - LastPubReliableSeq uint32 - Country string - PreferVideoSizeFromMedia bool - UseSinglePeerConnection bool - EnableDataTracks bool - EnableRTPStreamRestartDetection bool - ForceBackupCodecPolicySimulcast bool - DisableTransceiverReuseForE2EE bool + PublishEnabledCodecs []*livekit.Codec + SubscribeEnabledCodecs []*livekit.Codec + Logger logger.Logger + LoggerResolver logger.DeferredFieldResolver + Reporter roomobs.ParticipantSessionReporter + ReporterResolver roomobs.ParticipantReporterResolver + SimTracks map[uint32]interceptor.SimulcastTrackInfo + Grants *auth.ClaimGrants + InitialVersion uint32 + ClientConf *livekit.ClientConfiguration + ClientInfo ClientInfo + Region string + Migration bool + Reconnect bool + AdaptiveStream bool + AllowTCPFallback bool + TCPFallbackRTTThreshold int + AllowUDPUnstableFallback bool + TURNSEnabled bool + ParticipantListener types.LocalParticipantListener + ParticipantHelper types.LocalParticipantHelper + DisableSupervisor bool + ReconnectOnPublicationError bool + ReconnectOnSubscriptionError bool + ReconnectOnDataChannelError bool + VersionGenerator utils.TimedVersionGenerator + DisableDynacast bool + SubscriberAllowPause bool + SubscriptionLimitAudio int32 + SubscriptionLimitVideo int32 + PlayoutDelay *livekit.PlayoutDelay + SyncStreams bool + ForwardStats *sfu.ForwardStats + DisableSenderReportPassThrough bool + MetricConfig metric.MetricConfig + UseOneShotSignallingMode bool + EnableMetrics bool + DataChannelMaxBufferedAmount uint64 + DatachannelSlowThreshold int + DatachannelLossyTargetLatency time.Duration + FireOnTrackBySdp bool + DisableCodecRegression bool + LastPubReliableSeq uint32 + Country string + PreferVideoSizeFromMedia bool + UseSinglePeerConnection bool + EnableDataTracks bool + EnableRTPStreamRestartDetection bool + ForceBackupCodecPolicySimulcast bool + DisableTransceiverReuseForE2EE bool + EnableParticipantAsyncAttributes bool } type ParticipantImpl struct { @@ -332,6 +333,8 @@ type ParticipantImpl struct { rpcLock sync.Mutex rpcPendingAcks map[string]*utils.DataChannelRpcPendingAckHandler rpcPendingResponses map[string]*utils.DataChannelRpcPendingResponseHandler + + asyncAttributes *ParticipantAsyncAttributes } func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { @@ -370,6 +373,9 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { telemetryGuard: &telemetry.ReferenceGuard{}, nextSubscribedDataTrackHandle: uint16(rand.Intn(256)), requireBroadcast: params.Grants.Metadata != "" || len(params.Grants.Attributes) != 0, + asyncAttributes: NewParticipantAsyncAttributes(ParticipantAsyncAttributesParams{ + Logger: params.Logger, + }), } p.setupSignalling() diff --git a/pkg/rtc/participant_async_attributes.go b/pkg/rtc/participant_async_attributes.go new file mode 100644 index 000000000..aa546f46e --- /dev/null +++ b/pkg/rtc/participant_async_attributes.go @@ -0,0 +1,105 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package rtc + +import ( + "fmt" + "strconv" + "strings" + "sync" + + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" +) + +type ParticipantAsyncAttributesParams struct { + Logger logger.Logger +} + +type ParticipantAsyncAttributes struct { + params ParticipantAsyncAttributesParams + lock sync.Mutex + attributes map[string][]byte +} + +func NewParticipantAsyncAttributes(params ParticipantAsyncAttributesParams) *ParticipantAsyncAttributes { + return &ParticipantAsyncAttributes{ + params: params, + attributes: make(map[string][]byte), + } +} + +func (p *ParticipantAsyncAttributes) Add(id *livekit.DataTrackSchemaId, value []byte) { + p.lock.Lock() + defer p.lock.Unlock() + + if id == nil { + return + } + + p.attributes[ToParticipantAsyncAttributeKey(id)] = value +} + +func (p *ParticipantAsyncAttributes) Delete(id *livekit.DataTrackSchemaId) { + p.lock.Lock() + defer p.lock.Unlock() + + if id == nil { + return + } + + delete(p.attributes, ToParticipantAsyncAttributeKey(id)) +} + +func (p *ParticipantAsyncAttributes) Get(id *livekit.DataTrackSchemaId) *livekit.DataTrackSchemaDefinition { + p.lock.Lock() + defer p.lock.Unlock() + + if id == nil { + return nil + } + + value, ok := p.attributes[ToParticipantAsyncAttributeKey(id)] + if !ok { + return nil + } + + return &livekit.DataTrackSchemaDefinition{ + Id: id, + Definition: value, + } +} + +func (p *ParticipantAsyncAttributes) GetAll() map[string][]byte { + p.lock.Lock() + defer p.lock.Unlock() + + return p.attributes +} + +// ------------------------------- + +func ToParticipantAsyncAttributeKey(id *livekit.DataTrackSchemaId) string { + return fmt.Sprintf("%s/%d", id.Name, id.Encoding) +} + +func FromParticipantAsyncAttributeKey(key string) *livekit.DataTrackSchemaId { + parts := strings.Split(key, "/") + encoding, _ := strconv.Atoi(parts[1]) + return &livekit.DataTrackSchemaId{ + Name: parts[0], + Encoding: livekit.DataTrackSchemaEncoding(encoding), + } +} diff --git a/pkg/rtc/participant_async_attributes_handler.go b/pkg/rtc/participant_async_attributes_handler.go new file mode 100644 index 000000000..993f81635 --- /dev/null +++ b/pkg/rtc/participant_async_attributes_handler.go @@ -0,0 +1,126 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package rtc + +import ( + "github.com/livekit/livekit-server/pkg/rtc/types" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" + "github.com/livekit/protocol/utils" +) + +func (p *ParticipantImpl) HandleDefineDataTrackSchemaRequest(req *livekit.DefineDataTrackSchemaRequest) { + if !p.params.EnableParticipantAsyncAttributes { + p.pubLogger.Warnw("async attributes not enabled", nil, "req", logger.Proto(req)) + p.sendRequestResponse(&livekit.RequestResponse{ + Reason: livekit.RequestResponse_NOT_ALLOWED, + Message: "async attributes not enabled", + Request: &livekit.RequestResponse_DefineDataTrackSchema{ + DefineDataTrackSchema: utils.CloneProto(req), + }, + }) + return + } + + if req.SchemaDefinition == nil || req.SchemaDefinition.Id == nil || len(req.SchemaDefinition.Id.Name) == 0 || len(req.SchemaDefinition.Id.Name) > 256 { + p.pubLogger.Warnw("aync attribute definition is invalid", nil, "req", logger.Proto(req)) + p.sendRequestResponse(&livekit.RequestResponse{ + Reason: livekit.RequestResponse_INVALID_REQUEST, + Message: "async attribute definition is invalid", + Request: &livekit.RequestResponse_DefineDataTrackSchema{ + DefineDataTrackSchema: utils.CloneProto(req), + }, + }) + return + } + + if len(req.SchemaDefinition.Definition) == 0 { + p.sendRequestResponse(&livekit.RequestResponse{ + Reason: livekit.RequestResponse_INVALID_REQUEST, + Message: "async attribute definition is empty", + Request: &livekit.RequestResponse_DefineDataTrackSchema{ + DefineDataTrackSchema: utils.CloneProto(req), + }, + }) + return + } + + if !p.params.LimitConfig.CanAddAsyncAttribute( + p.asyncAttributes.GetAll(), + ToParticipantAsyncAttributeKey(req.SchemaDefinition.Id), + req.SchemaDefinition.Definition, + ) { + p.sendRequestResponse(&livekit.RequestResponse{ + Reason: livekit.RequestResponse_LIMIT_EXCEEDED, + Message: "async attribute definition exceeds limit", + Request: &livekit.RequestResponse_DefineDataTrackSchema{ + DefineDataTrackSchema: utils.CloneProto(req), + }, + }) + return + } + + p.AddDataTrackSchema(req.SchemaDefinition) +} + +func (p *ParticipantImpl) HandleGetDataTrackSchemaRequest(req *livekit.GetDataTrackSchemaRequest) { + if req.SchemaId == nil { + p.sendRequestResponse(&livekit.RequestResponse{ + Reason: livekit.RequestResponse_INVALID_REQUEST, + Message: "async attribute id is required", + Request: &livekit.RequestResponse_GetDataTrackSchema{ + GetDataTrackSchema: utils.CloneProto(req), + }, + }) + return + } + + p.listener().OnGetDataTrackSchema(p, req) +} + +func (p *ParticipantImpl) AddDataTrackSchema(definition *livekit.DataTrackSchemaDefinition) { + p.asyncAttributes.Add(definition.Id, definition.Definition) +} + +func (p *ParticipantImpl) GetDataTrackSchema(id *livekit.DataTrackSchemaId) *livekit.DataTrackSchemaDefinition { + return p.asyncAttributes.Get(id) +} + +func (p *ParticipantImpl) ProcessGetDataTrackSchemaRequest(req *livekit.GetDataTrackSchemaRequest, publisher types.Participant) { + if publisher == nil { + p.sendRequestResponse(&livekit.RequestResponse{ + Reason: livekit.RequestResponse_NOT_FOUND, + Message: "participant not found", + Request: &livekit.RequestResponse_GetDataTrackSchema{ + GetDataTrackSchema: utils.CloneProto(req), + }, + }) + return + } + + asyncAttribute := publisher.GetDataTrackSchema(req.SchemaId) + if asyncAttribute == nil { + p.sendRequestResponse(&livekit.RequestResponse{ + Reason: livekit.RequestResponse_NOT_FOUND, + Message: "async attribute not found", + Request: &livekit.RequestResponse_GetDataTrackSchema{ + GetDataTrackSchema: utils.CloneProto(req), + }, + }) + return + } + + p.sendGetDataTrackSchemaResponse(asyncAttribute) +} diff --git a/pkg/rtc/participant_async_attributes_handler_test.go b/pkg/rtc/participant_async_attributes_handler_test.go new file mode 100644 index 000000000..9d8201c12 --- /dev/null +++ b/pkg/rtc/participant_async_attributes_handler_test.go @@ -0,0 +1,319 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package rtc + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/livekit/protocol/livekit" + + "github.com/livekit/livekit-server/pkg/config" + "github.com/livekit/livekit-server/pkg/routing/routingfakes" + "github.com/livekit/livekit-server/pkg/rtc/types/typesfakes" +) + +func newParticipantWithAsyncAttributes(t *testing.T, enabled bool, maxSize uint32) *ParticipantImpl { + t.Helper() + p := newParticipantForTest("test") + p.params.EnableParticipantAsyncAttributes = enabled + p.params.LimitConfig = config.LimitConfig{ + MaxAsyncAttributesSize: maxSize, + } + return p +} + +func lastRequestResponse(t *testing.T, sink *routingfakes.FakeMessageSink, idx int) *livekit.RequestResponse { + t.Helper() + msg := sink.WriteMessageArgsForCall(idx).(*livekit.SignalResponse) + rr, ok := msg.Message.(*livekit.SignalResponse_RequestResponse) + require.True(t, ok, "expected SignalResponse_RequestResponse, got %T", msg.Message) + return rr.RequestResponse +} + +func TestHandleDefineDataTrackSchemaRequest(t *testing.T) { + t.Run("returns NOT_ALLOWED when feature not enabled", func(t *testing.T) { + p := newParticipantWithAsyncAttributes(t, false, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + req := &livekit.DefineDataTrackSchemaRequest{ + SchemaDefinition: &livekit.DataTrackSchemaDefinition{ + Id: &livekit.DataTrackSchemaId{ + Name: "schema-1", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + }, + Definition: []byte("def"), + }, + } + p.HandleDefineDataTrackSchemaRequest(req) + + require.Equal(t, 1, sink.WriteMessageCallCount()) + rr := lastRequestResponse(t, sink, 0) + require.Equal(t, livekit.RequestResponse_NOT_ALLOWED, rr.Reason) + require.NotNil(t, rr.GetDefineDataTrackSchema()) + require.Empty(t, p.asyncAttributes.GetAll()) + }) + + t.Run("returns INVALID_REQUEST when schema definition is nil", func(t *testing.T) { + p := newParticipantWithAsyncAttributes(t, true, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + p.HandleDefineDataTrackSchemaRequest(&livekit.DefineDataTrackSchemaRequest{}) + + require.Equal(t, 1, sink.WriteMessageCallCount()) + rr := lastRequestResponse(t, sink, 0) + require.Equal(t, livekit.RequestResponse_INVALID_REQUEST, rr.Reason) + require.Empty(t, p.asyncAttributes.GetAll()) + }) + + t.Run("returns INVALID_REQUEST when id is nil", func(t *testing.T) { + p := newParticipantWithAsyncAttributes(t, true, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + p.HandleDefineDataTrackSchemaRequest(&livekit.DefineDataTrackSchemaRequest{ + SchemaDefinition: &livekit.DataTrackSchemaDefinition{ + Definition: []byte("def"), + }, + }) + + require.Equal(t, 1, sink.WriteMessageCallCount()) + rr := lastRequestResponse(t, sink, 0) + require.Equal(t, livekit.RequestResponse_INVALID_REQUEST, rr.Reason) + }) + + t.Run("returns INVALID_REQUEST when name is empty", func(t *testing.T) { + p := newParticipantWithAsyncAttributes(t, true, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + p.HandleDefineDataTrackSchemaRequest(&livekit.DefineDataTrackSchemaRequest{ + SchemaDefinition: &livekit.DataTrackSchemaDefinition{ + Id: &livekit.DataTrackSchemaId{ + Name: "", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + }, + Definition: []byte("def"), + }, + }) + + require.Equal(t, 1, sink.WriteMessageCallCount()) + rr := lastRequestResponse(t, sink, 0) + require.Equal(t, livekit.RequestResponse_INVALID_REQUEST, rr.Reason) + }) + + t.Run("returns INVALID_REQUEST when name exceeds 256 chars", func(t *testing.T) { + p := newParticipantWithAsyncAttributes(t, true, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + p.HandleDefineDataTrackSchemaRequest(&livekit.DefineDataTrackSchemaRequest{ + SchemaDefinition: &livekit.DataTrackSchemaDefinition{ + Id: &livekit.DataTrackSchemaId{ + Name: strings.Repeat("a", 257), + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + }, + Definition: []byte("def"), + }, + }) + + require.Equal(t, 1, sink.WriteMessageCallCount()) + rr := lastRequestResponse(t, sink, 0) + require.Equal(t, livekit.RequestResponse_INVALID_REQUEST, rr.Reason) + }) + + t.Run("returns INVALID_REQUEST when definition is empty", func(t *testing.T) { + p := newParticipantWithAsyncAttributes(t, true, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + p.HandleDefineDataTrackSchemaRequest(&livekit.DefineDataTrackSchemaRequest{ + SchemaDefinition: &livekit.DataTrackSchemaDefinition{ + Id: &livekit.DataTrackSchemaId{ + Name: "schema-1", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + }, + Definition: nil, + }, + }) + + require.Equal(t, 1, sink.WriteMessageCallCount()) + rr := lastRequestResponse(t, sink, 0) + require.Equal(t, livekit.RequestResponse_INVALID_REQUEST, rr.Reason) + require.Empty(t, p.asyncAttributes.GetAll()) + }) + + t.Run("returns LIMIT_EXCEEDED when adding would breach the limit", func(t *testing.T) { + p := newParticipantWithAsyncAttributes(t, true, 16) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + p.HandleDefineDataTrackSchemaRequest(&livekit.DefineDataTrackSchemaRequest{ + SchemaDefinition: &livekit.DataTrackSchemaDefinition{ + Id: &livekit.DataTrackSchemaId{ + Name: "schema-1", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + }, + Definition: []byte(strings.Repeat("x", 32)), + }, + }) + + require.Equal(t, 1, sink.WriteMessageCallCount()) + rr := lastRequestResponse(t, sink, 0) + require.Equal(t, livekit.RequestResponse_LIMIT_EXCEEDED, rr.Reason) + require.Empty(t, p.asyncAttributes.GetAll()) + }) + + t.Run("stores a valid definition and sends no response", func(t *testing.T) { + p := newParticipantWithAsyncAttributes(t, true, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + id := &livekit.DataTrackSchemaId{ + Name: "schema-1", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + } + def := []byte("definition-bytes") + + p.HandleDefineDataTrackSchemaRequest(&livekit.DefineDataTrackSchemaRequest{ + SchemaDefinition: &livekit.DataTrackSchemaDefinition{ + Id: id, + Definition: def, + }, + }) + + // on success no response is sent to the client + require.Equal(t, 0, sink.WriteMessageCallCount()) + + stored := p.asyncAttributes.Get(id) + require.NotNil(t, stored) + require.Equal(t, def, stored.Definition) + }) +} + +func TestHandleGetDataTrackSchemaRequest(t *testing.T) { + t.Run("returns INVALID_REQUEST when schema id is missing", func(t *testing.T) { + p := newParticipantWithAsyncAttributes(t, true, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + p.HandleGetDataTrackSchemaRequest(&livekit.GetDataTrackSchemaRequest{ + ParticipantIdentity: "other", + }) + + require.Equal(t, 1, sink.WriteMessageCallCount()) + rr := lastRequestResponse(t, sink, 0) + require.Equal(t, livekit.RequestResponse_INVALID_REQUEST, rr.Reason) + }) + + t.Run("forwards request to listener when schema id is provided", func(t *testing.T) { + p := newParticipantWithAsyncAttributes(t, true, 0) + listener := p.params.ParticipantListener.(*typesfakes.FakeLocalParticipantListener) + + req := &livekit.GetDataTrackSchemaRequest{ + ParticipantIdentity: "other", + SchemaId: &livekit.DataTrackSchemaId{ + Name: "schema-1", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + }, + } + p.HandleGetDataTrackSchemaRequest(req) + + require.Equal(t, 1, listener.OnGetDataTrackSchemaCallCount()) + gotParticipant, gotReq := listener.OnGetDataTrackSchemaArgsForCall(0) + require.Equal(t, p, gotParticipant) + require.Equal(t, req, gotReq) + }) +} + +func TestGetDataTrackSchema(t *testing.T) { + p := newParticipantWithAsyncAttributes(t, true, 0) + + id := &livekit.DataTrackSchemaId{ + Name: "schema-1", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + } + require.Nil(t, p.GetDataTrackSchema(id)) + + p.asyncAttributes.Add(id, []byte("definition")) + got := p.GetDataTrackSchema(id) + require.NotNil(t, got) + require.Equal(t, id.Name, got.Id.Name) + require.Equal(t, []byte("definition"), got.Definition) +} + +func TestProcessGetDataTrackSchemaRequest(t *testing.T) { + t.Run("returns NOT_FOUND when publisher is nil", func(t *testing.T) { + p := newParticipantWithAsyncAttributes(t, true, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + p.ProcessGetDataTrackSchemaRequest(&livekit.GetDataTrackSchemaRequest{ + SchemaId: &livekit.DataTrackSchemaId{ + Name: "schema-1", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + }, + }, nil) + + require.Equal(t, 1, sink.WriteMessageCallCount()) + rr := lastRequestResponse(t, sink, 0) + require.Equal(t, livekit.RequestResponse_NOT_FOUND, rr.Reason) + require.Contains(t, rr.Message, "participant") + }) + + t.Run("returns NOT_FOUND when publisher has no matching schema", func(t *testing.T) { + p := newParticipantWithAsyncAttributes(t, true, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + publisher := &typesfakes.FakeParticipant{} + publisher.GetDataTrackSchemaReturns(nil) + + req := &livekit.GetDataTrackSchemaRequest{ + SchemaId: &livekit.DataTrackSchemaId{ + Name: "schema-1", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + }, + } + p.ProcessGetDataTrackSchemaRequest(req, publisher) + + require.Equal(t, 1, publisher.GetDataTrackSchemaCallCount()) + require.Equal(t, req.SchemaId, publisher.GetDataTrackSchemaArgsForCall(0)) + + require.Equal(t, 1, sink.WriteMessageCallCount()) + rr := lastRequestResponse(t, sink, 0) + require.Equal(t, livekit.RequestResponse_NOT_FOUND, rr.Reason) + }) + + t.Run("sends schema response when publisher has a matching schema", func(t *testing.T) { + p := newParticipantWithAsyncAttributes(t, true, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + id := &livekit.DataTrackSchemaId{ + Name: "schema-1", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + } + def := &livekit.DataTrackSchemaDefinition{ + Id: id, + Definition: []byte("definition-bytes"), + } + + publisher := &typesfakes.FakeParticipant{} + publisher.GetDataTrackSchemaReturns(def) + + p.ProcessGetDataTrackSchemaRequest(&livekit.GetDataTrackSchemaRequest{ + SchemaId: id, + }, publisher) + + require.Equal(t, 1, sink.WriteMessageCallCount()) + msg := sink.WriteMessageArgsForCall(0).(*livekit.SignalResponse) + response, ok := msg.Message.(*livekit.SignalResponse_GetDataTrackSchemaResponse) + require.True(t, ok, "expected SignalResponse_GetDataTrackSchemaResponse, got %T", msg.Message) + require.Equal(t, def, response.GetDataTrackSchemaResponse.SchemaDefinition) + }) +} diff --git a/pkg/rtc/participant_async_attributes_test.go b/pkg/rtc/participant_async_attributes_test.go new file mode 100644 index 000000000..370fadb91 --- /dev/null +++ b/pkg/rtc/participant_async_attributes_test.go @@ -0,0 +1,207 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package rtc + +import ( + "sync" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" +) + +func newTestAsyncAttributes() *ParticipantAsyncAttributes { + return NewParticipantAsyncAttributes(ParticipantAsyncAttributesParams{ + Logger: logger.GetLogger(), + }) +} + +func TestParticipantAsyncAttributes_AddAndGet(t *testing.T) { + a := newTestAsyncAttributes() + + id := &livekit.DataTrackSchemaId{ + Name: "schema-1", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + } + value := []byte("definition-bytes") + + a.Add(id, value) + + got := a.Get(id) + require.NotNil(t, got) + require.Equal(t, id.Name, got.Id.Name) + require.Equal(t, id.Encoding, got.Id.Encoding) + require.Equal(t, value, got.Definition) +} + +func TestParticipantAsyncAttributes_AddOverwrites(t *testing.T) { + a := newTestAsyncAttributes() + + id := &livekit.DataTrackSchemaId{ + Name: "schema-1", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + } + a.Add(id, []byte("v1")) + a.Add(id, []byte("v2")) + + got := a.Get(id) + require.NotNil(t, got) + require.Equal(t, []byte("v2"), got.Definition) + + require.Len(t, a.GetAll(), 1) +} + +func TestParticipantAsyncAttributes_DifferentEncodingsAreDistinct(t *testing.T) { + a := newTestAsyncAttributes() + + idProto := &livekit.DataTrackSchemaId{ + Name: "schema-1", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + } + idJSON := &livekit.DataTrackSchemaId{ + Name: "schema-1", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_JSON_SCHEMA, + } + + a.Add(idProto, []byte("proto-def")) + a.Add(idJSON, []byte("json-def")) + + gotProto := a.Get(idProto) + require.NotNil(t, gotProto) + require.Equal(t, []byte("proto-def"), gotProto.Definition) + + gotJSON := a.Get(idJSON) + require.NotNil(t, gotJSON) + require.Equal(t, []byte("json-def"), gotJSON.Definition) + + require.Len(t, a.GetAll(), 2) +} + +func TestParticipantAsyncAttributes_Delete(t *testing.T) { + a := newTestAsyncAttributes() + + id := &livekit.DataTrackSchemaId{ + Name: "schema-1", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + } + a.Add(id, []byte("definition")) + + a.Delete(id) + require.Nil(t, a.Get(id)) + require.Empty(t, a.GetAll()) + + // deleting a non-existent key is a no-op + a.Delete(id) + require.Empty(t, a.GetAll()) +} + +func TestParticipantAsyncAttributes_NilId(t *testing.T) { + a := newTestAsyncAttributes() + + // nil id should be silently ignored, not panic + a.Add(nil, []byte("definition")) + require.Empty(t, a.GetAll()) + + require.Nil(t, a.Get(nil)) + + a.Delete(nil) + require.Empty(t, a.GetAll()) +} + +func TestParticipantAsyncAttributes_GetMissing(t *testing.T) { + a := newTestAsyncAttributes() + + id := &livekit.DataTrackSchemaId{ + Name: "missing", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + } + require.Nil(t, a.Get(id)) +} + +func TestParticipantAsyncAttributes_GetAllContents(t *testing.T) { + a := newTestAsyncAttributes() + + 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, + } + + a.Add(id1, []byte("def-1")) + a.Add(id2, []byte("def-2")) + + all := a.GetAll() + require.Len(t, all, 2) + require.Equal(t, []byte("def-1"), all[ToParticipantAsyncAttributeKey(id1)]) + require.Equal(t, []byte("def-2"), all[ToParticipantAsyncAttributeKey(id2)]) +} + +func TestParticipantAsyncAttributes_ConcurrentAccess(t *testing.T) { + a := newTestAsyncAttributes() + + const numGoroutines = 16 + const opsPerGoroutine = 100 + + var wg sync.WaitGroup + wg.Add(numGoroutines) + for g := 0; g < numGoroutines; g++ { + go func(g int) { + defer wg.Done() + for i := 0; i < opsPerGoroutine; i++ { + id := &livekit.DataTrackSchemaId{ + Name: "schema", + Encoding: livekit.DataTrackSchemaEncoding(g % 8), + } + a.Add(id, []byte("v")) + _ = a.Get(id) + _ = a.GetAll() + if i%3 == 0 { + a.Delete(id) + } + } + }(g) + } + wg.Wait() +} + +func TestToKeyFromKey(t *testing.T) { + ids := []*livekit.DataTrackSchemaId{ + { + Name: "schema-1", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + }, + { + Name: "another-schema", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_JSON_SCHEMA, + }, + { + Name: "x", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_UNSPECIFIED, + }, + } + for _, id := range ids { + t.Run(id.Name, func(t *testing.T) { + key := ToParticipantAsyncAttributeKey(id) + roundTripped := FromParticipantAsyncAttributeKey(key) + require.Equal(t, id.Name, roundTripped.Name) + require.Equal(t, id.Encoding, roundTripped.Encoding) + }) + } +} diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go index 3fa388adc..14b3a8a2c 100644 --- a/pkg/rtc/participant_signal.go +++ b/pkg/rtc/participant_signal.go @@ -368,3 +368,9 @@ func (p *ParticipantImpl) SendDataTrackSubscriberHandles(handles map[uint32]*liv SubHandles: handles, })) } + +func (p *ParticipantImpl) sendGetDataTrackSchemaResponse(definition *livekit.DataTrackSchemaDefinition) error { + return p.signaller.WriteMessage(p.signalling.SignalGetDataTrackSchemaResponse(&livekit.GetDataTrackSchemaResponse{ + SchemaDefinition: definition, + })) +} diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 8a3118dcb..d504d33ad 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -1374,6 +1374,11 @@ func (r *Room) onUpdateDataSubscriptions(participant types.LocalParticipant, req } } +func (r *Room) onGetDataTrackSchema(participant types.LocalParticipant, req *livekit.GetDataTrackSchemaRequest) { + publisher := r.GetParticipant(livekit.ParticipantIdentity(req.ParticipantIdentity)) + participant.ProcessGetDataTrackSchemaRequest(req, publisher) +} + func (r *Room) onLeave(p types.LocalParticipant, reason types.ParticipantCloseReason) { r.RemoveParticipant(p.Identity(), p.ID(), reason) } @@ -1983,6 +1988,13 @@ func (l *localParticipantListener) OnUpdateDataSubscriptions(p types.LocalPartic l.room.onUpdateDataSubscriptions(p, req) } +func (l *localParticipantListener) OnDefineDataTrackSchema(_p types.LocalParticipant, _definition *livekit.DataTrackSchemaDefinition) { +} + +func (l *localParticipantListener) OnGetDataTrackSchema(p types.LocalParticipant, req *livekit.GetDataTrackSchemaRequest) { + l.room.onGetDataTrackSchema(p, req) +} + func (l *localParticipantListener) OnSyncState(p types.LocalParticipant, state *livekit.SyncState) error { return l.room.onSyncState(p, state) } diff --git a/pkg/rtc/signalling/interfaces.go b/pkg/rtc/signalling/interfaces.go index 2470e9d31..d6ed03c16 100644 --- a/pkg/rtc/signalling/interfaces.go +++ b/pkg/rtc/signalling/interfaces.go @@ -62,4 +62,5 @@ type ParticipantSignalling interface { SignalPublishDataTrackResponse(publishDataTrackResponse *livekit.PublishDataTrackResponse) proto.Message SignalUnpublishDataTrackResponse(unpublishDataTrackResponse *livekit.UnpublishDataTrackResponse) proto.Message SignalDataTrackSubscriberHandles(dataTrackSubscriberHandles *livekit.DataTrackSubscriberHandles) proto.Message + SignalGetDataTrackSchemaResponse(getDataTrackSchemaResponse *livekit.GetDataTrackSchemaResponse) proto.Message } diff --git a/pkg/rtc/signalling/signalhandler.go b/pkg/rtc/signalling/signalhandler.go index 7e69aeee2..0b3b282a8 100644 --- a/pkg/rtc/signalling/signalhandler.go +++ b/pkg/rtc/signalling/signalhandler.go @@ -151,6 +151,12 @@ func (s *signalhandler) HandleMessage(msg proto.Message) error { case *livekit.SignalRequest_UpdateDataSubscription: s.params.Participant.HandleUpdateDataSubscription(msg.UpdateDataSubscription) + + case *livekit.SignalRequest_DefineDataTrackSchema: + s.params.Participant.HandleDefineDataTrackSchemaRequest(msg.DefineDataTrackSchema) + + case *livekit.SignalRequest_GetDataTrackSchema: + s.params.Participant.HandleGetDataTrackSchemaRequest(msg.GetDataTrackSchema) } return nil diff --git a/pkg/rtc/signalling/signalling.go b/pkg/rtc/signalling/signalling.go index fffc2ba29..946ac9914 100644 --- a/pkg/rtc/signalling/signalling.go +++ b/pkg/rtc/signalling/signalling.go @@ -258,3 +258,11 @@ func (s *signalling) SignalDataTrackSubscriberHandles(dataTrackSubscriberHandles }, } } + +func (u *signalling) SignalGetDataTrackSchemaResponse(getDataTrackSchemaResponse *livekit.GetDataTrackSchemaResponse) proto.Message { + return &livekit.SignalResponse{ + Message: &livekit.SignalResponse_GetDataTrackSchemaResponse{ + GetDataTrackSchemaResponse: getDataTrackSchemaResponse, + }, + } +} diff --git a/pkg/rtc/signalling/signallingunimplemented.go b/pkg/rtc/signalling/signallingunimplemented.go index dca48779f..b229dff63 100644 --- a/pkg/rtc/signalling/signallingunimplemented.go +++ b/pkg/rtc/signalling/signallingunimplemented.go @@ -127,3 +127,7 @@ func (u *signallingUnimplemented) SignalUnpublishDataTrackResponse(unpublishData func (u *signallingUnimplemented) SignalDataTrackSubscriberHandles(dataTrackSubscriberHandles *livekit.DataTrackSubscriberHandles) proto.Message { return nil } + +func (u *signallingUnimplemented) SignalGetDataTrackSchemaResponse(getDataTrackSchemaResponse *livekit.GetDataTrackSchemaResponse) proto.Message { + return nil +} diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index e77393e80..9a07cc823 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -355,6 +355,9 @@ type Participant interface { HandleReceivedDataTrackMessage([]byte, *datatrack.Packet, int64) GetParticipantListener() ParticipantListener + + AddDataTrackSchema(definition *livekit.DataTrackSchemaDefinition) + GetDataTrackSchema(id *livekit.DataTrackSchemaId) *livekit.DataTrackSchemaDefinition } // ------------------------------------------------------- @@ -560,6 +563,9 @@ type LocalParticipant interface { HandlePublishDataTrackRequest(*livekit.PublishDataTrackRequest) HandleUnpublishDataTrackRequest(*livekit.UnpublishDataTrackRequest) HandleUpdateDataSubscription(*livekit.UpdateDataSubscription) + HandleDefineDataTrackSchemaRequest(*livekit.DefineDataTrackSchemaRequest) + HandleGetDataTrackSchemaRequest(*livekit.GetDataTrackSchemaRequest) + ProcessGetDataTrackSchemaRequest(*livekit.GetDataTrackSchemaRequest, Participant) HandleSignalMessage(msg proto.Message) error @@ -619,6 +625,8 @@ type LocalParticipantListener interface { ) OnUpdateSubscriptionPermission(LocalParticipant, *livekit.SubscriptionPermission) error OnUpdateDataSubscriptions(LocalParticipant, *livekit.UpdateDataSubscription) + OnDefineDataTrackSchema(LocalParticipant, *livekit.DataTrackSchemaDefinition) + OnGetDataTrackSchema(LocalParticipant, *livekit.GetDataTrackSchemaRequest) OnSyncState(LocalParticipant, *livekit.SyncState) error OnSimulateScenario(LocalParticipant, *livekit.SimulateScenario) error OnLeave(LocalParticipant, ParticipantCloseReason) @@ -650,6 +658,10 @@ func (*NullLocalParticipantListener) OnUpdateSubscriptionPermission(LocalPartici } func (*NullLocalParticipantListener) OnUpdateDataSubscriptions(LocalParticipant, *livekit.UpdateDataSubscription) { } +func (*NullLocalParticipantListener) OnDefineDataTrackSchema(LocalParticipant, *livekit.DataTrackSchemaDefinition) { +} +func (*NullLocalParticipantListener) OnGetDataTrackSchema(LocalParticipant, *livekit.GetDataTrackSchemaRequest) { +} func (*NullLocalParticipantListener) OnSyncState(LocalParticipant, *livekit.SyncState) error { return nil } diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 96686a0d1..b05c9983b 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -33,6 +33,11 @@ type FakeLocalParticipant struct { activeAtReturnsOnCall map[int]struct { result1 time.Time } + AddDataTrackSchemaStub func(*livekit.DataTrackSchemaDefinition) + addDataTrackSchemaMutex sync.RWMutex + addDataTrackSchemaArgsForCall []struct { + arg1 *livekit.DataTrackSchemaDefinition + } AddOnCloseStub func(string, func(types.LocalParticipant)) addOnCloseMutex sync.RWMutex addOnCloseArgsForCall []struct { @@ -305,6 +310,17 @@ type FakeLocalParticipant struct { getCountryReturnsOnCall map[int]struct { result1 string } + GetDataTrackSchemaStub func(*livekit.DataTrackSchemaId) *livekit.DataTrackSchemaDefinition + getDataTrackSchemaMutex sync.RWMutex + getDataTrackSchemaArgsForCall []struct { + arg1 *livekit.DataTrackSchemaId + } + getDataTrackSchemaReturns struct { + result1 *livekit.DataTrackSchemaDefinition + } + getDataTrackSchemaReturnsOnCall map[int]struct { + result1 *livekit.DataTrackSchemaDefinition + } GetDataTrackTransportStub func() types.DataTrackTransport getDataTrackTransportMutex sync.RWMutex getDataTrackTransportArgsForCall []struct { @@ -566,6 +582,16 @@ type FakeLocalParticipant struct { handleAnswerArgsForCall []struct { arg1 *livekit.SessionDescription } + HandleDefineDataTrackSchemaRequestStub func(*livekit.DefineDataTrackSchemaRequest) + handleDefineDataTrackSchemaRequestMutex sync.RWMutex + handleDefineDataTrackSchemaRequestArgsForCall []struct { + arg1 *livekit.DefineDataTrackSchemaRequest + } + HandleGetDataTrackSchemaRequestStub func(*livekit.GetDataTrackSchemaRequest) + handleGetDataTrackSchemaRequestMutex sync.RWMutex + handleGetDataTrackSchemaRequestArgsForCall []struct { + arg1 *livekit.GetDataTrackSchemaRequest + } HandleICERestartSDPFragmentStub func(string) (string, error) handleICERestartSDPFragmentMutex sync.RWMutex handleICERestartSDPFragmentArgsForCall []struct { @@ -976,6 +1002,12 @@ type FakeLocalParticipant struct { arg2 chan string arg3 chan error } + ProcessGetDataTrackSchemaRequestStub func(*livekit.GetDataTrackSchemaRequest, types.Participant) + processGetDataTrackSchemaRequestMutex sync.RWMutex + processGetDataTrackSchemaRequestArgsForCall []struct { + arg1 *livekit.GetDataTrackSchemaRequest + arg2 types.Participant + } ProtocolVersionStub func() types.ProtocolVersion protocolVersionMutex sync.RWMutex protocolVersionArgsForCall []struct { @@ -1574,6 +1606,38 @@ func (fake *FakeLocalParticipant) ActiveAtReturnsOnCall(i int, result1 time.Time }{result1} } +func (fake *FakeLocalParticipant) AddDataTrackSchema(arg1 *livekit.DataTrackSchemaDefinition) { + fake.addDataTrackSchemaMutex.Lock() + fake.addDataTrackSchemaArgsForCall = append(fake.addDataTrackSchemaArgsForCall, struct { + arg1 *livekit.DataTrackSchemaDefinition + }{arg1}) + stub := fake.AddDataTrackSchemaStub + fake.recordInvocation("AddDataTrackSchema", []interface{}{arg1}) + fake.addDataTrackSchemaMutex.Unlock() + if stub != nil { + fake.AddDataTrackSchemaStub(arg1) + } +} + +func (fake *FakeLocalParticipant) AddDataTrackSchemaCallCount() int { + fake.addDataTrackSchemaMutex.RLock() + defer fake.addDataTrackSchemaMutex.RUnlock() + return len(fake.addDataTrackSchemaArgsForCall) +} + +func (fake *FakeLocalParticipant) AddDataTrackSchemaCalls(stub func(*livekit.DataTrackSchemaDefinition)) { + fake.addDataTrackSchemaMutex.Lock() + defer fake.addDataTrackSchemaMutex.Unlock() + fake.AddDataTrackSchemaStub = stub +} + +func (fake *FakeLocalParticipant) AddDataTrackSchemaArgsForCall(i int) *livekit.DataTrackSchemaDefinition { + fake.addDataTrackSchemaMutex.RLock() + defer fake.addDataTrackSchemaMutex.RUnlock() + argsForCall := fake.addDataTrackSchemaArgsForCall[i] + return argsForCall.arg1 +} + func (fake *FakeLocalParticipant) AddOnClose(arg1 string, arg2 func(types.LocalParticipant)) { fake.addOnCloseMutex.Lock() fake.addOnCloseArgsForCall = append(fake.addOnCloseArgsForCall, struct { @@ -2963,6 +3027,67 @@ func (fake *FakeLocalParticipant) GetCountryReturnsOnCall(i int, result1 string) }{result1} } +func (fake *FakeLocalParticipant) GetDataTrackSchema(arg1 *livekit.DataTrackSchemaId) *livekit.DataTrackSchemaDefinition { + fake.getDataTrackSchemaMutex.Lock() + ret, specificReturn := fake.getDataTrackSchemaReturnsOnCall[len(fake.getDataTrackSchemaArgsForCall)] + fake.getDataTrackSchemaArgsForCall = append(fake.getDataTrackSchemaArgsForCall, struct { + arg1 *livekit.DataTrackSchemaId + }{arg1}) + stub := fake.GetDataTrackSchemaStub + fakeReturns := fake.getDataTrackSchemaReturns + fake.recordInvocation("GetDataTrackSchema", []interface{}{arg1}) + fake.getDataTrackSchemaMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) GetDataTrackSchemaCallCount() int { + fake.getDataTrackSchemaMutex.RLock() + defer fake.getDataTrackSchemaMutex.RUnlock() + return len(fake.getDataTrackSchemaArgsForCall) +} + +func (fake *FakeLocalParticipant) GetDataTrackSchemaCalls(stub func(*livekit.DataTrackSchemaId) *livekit.DataTrackSchemaDefinition) { + fake.getDataTrackSchemaMutex.Lock() + defer fake.getDataTrackSchemaMutex.Unlock() + fake.GetDataTrackSchemaStub = stub +} + +func (fake *FakeLocalParticipant) GetDataTrackSchemaArgsForCall(i int) *livekit.DataTrackSchemaId { + fake.getDataTrackSchemaMutex.RLock() + defer fake.getDataTrackSchemaMutex.RUnlock() + argsForCall := fake.getDataTrackSchemaArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeLocalParticipant) GetDataTrackSchemaReturns(result1 *livekit.DataTrackSchemaDefinition) { + fake.getDataTrackSchemaMutex.Lock() + defer fake.getDataTrackSchemaMutex.Unlock() + fake.GetDataTrackSchemaStub = nil + fake.getDataTrackSchemaReturns = struct { + result1 *livekit.DataTrackSchemaDefinition + }{result1} +} + +func (fake *FakeLocalParticipant) GetDataTrackSchemaReturnsOnCall(i int, result1 *livekit.DataTrackSchemaDefinition) { + fake.getDataTrackSchemaMutex.Lock() + defer fake.getDataTrackSchemaMutex.Unlock() + fake.GetDataTrackSchemaStub = nil + if fake.getDataTrackSchemaReturnsOnCall == nil { + fake.getDataTrackSchemaReturnsOnCall = make(map[int]struct { + result1 *livekit.DataTrackSchemaDefinition + }) + } + fake.getDataTrackSchemaReturnsOnCall[i] = struct { + result1 *livekit.DataTrackSchemaDefinition + }{result1} +} + func (fake *FakeLocalParticipant) GetDataTrackTransport() types.DataTrackTransport { fake.getDataTrackTransportMutex.Lock() ret, specificReturn := fake.getDataTrackTransportReturnsOnCall[len(fake.getDataTrackTransportArgsForCall)] @@ -4355,6 +4480,70 @@ func (fake *FakeLocalParticipant) HandleAnswerArgsForCall(i int) *livekit.Sessio return argsForCall.arg1 } +func (fake *FakeLocalParticipant) HandleDefineDataTrackSchemaRequest(arg1 *livekit.DefineDataTrackSchemaRequest) { + fake.handleDefineDataTrackSchemaRequestMutex.Lock() + fake.handleDefineDataTrackSchemaRequestArgsForCall = append(fake.handleDefineDataTrackSchemaRequestArgsForCall, struct { + arg1 *livekit.DefineDataTrackSchemaRequest + }{arg1}) + stub := fake.HandleDefineDataTrackSchemaRequestStub + fake.recordInvocation("HandleDefineDataTrackSchemaRequest", []interface{}{arg1}) + fake.handleDefineDataTrackSchemaRequestMutex.Unlock() + if stub != nil { + fake.HandleDefineDataTrackSchemaRequestStub(arg1) + } +} + +func (fake *FakeLocalParticipant) HandleDefineDataTrackSchemaRequestCallCount() int { + fake.handleDefineDataTrackSchemaRequestMutex.RLock() + defer fake.handleDefineDataTrackSchemaRequestMutex.RUnlock() + return len(fake.handleDefineDataTrackSchemaRequestArgsForCall) +} + +func (fake *FakeLocalParticipant) HandleDefineDataTrackSchemaRequestCalls(stub func(*livekit.DefineDataTrackSchemaRequest)) { + fake.handleDefineDataTrackSchemaRequestMutex.Lock() + defer fake.handleDefineDataTrackSchemaRequestMutex.Unlock() + fake.HandleDefineDataTrackSchemaRequestStub = stub +} + +func (fake *FakeLocalParticipant) HandleDefineDataTrackSchemaRequestArgsForCall(i int) *livekit.DefineDataTrackSchemaRequest { + fake.handleDefineDataTrackSchemaRequestMutex.RLock() + defer fake.handleDefineDataTrackSchemaRequestMutex.RUnlock() + argsForCall := fake.handleDefineDataTrackSchemaRequestArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeLocalParticipant) HandleGetDataTrackSchemaRequest(arg1 *livekit.GetDataTrackSchemaRequest) { + fake.handleGetDataTrackSchemaRequestMutex.Lock() + fake.handleGetDataTrackSchemaRequestArgsForCall = append(fake.handleGetDataTrackSchemaRequestArgsForCall, struct { + arg1 *livekit.GetDataTrackSchemaRequest + }{arg1}) + stub := fake.HandleGetDataTrackSchemaRequestStub + fake.recordInvocation("HandleGetDataTrackSchemaRequest", []interface{}{arg1}) + fake.handleGetDataTrackSchemaRequestMutex.Unlock() + if stub != nil { + fake.HandleGetDataTrackSchemaRequestStub(arg1) + } +} + +func (fake *FakeLocalParticipant) HandleGetDataTrackSchemaRequestCallCount() int { + fake.handleGetDataTrackSchemaRequestMutex.RLock() + defer fake.handleGetDataTrackSchemaRequestMutex.RUnlock() + return len(fake.handleGetDataTrackSchemaRequestArgsForCall) +} + +func (fake *FakeLocalParticipant) HandleGetDataTrackSchemaRequestCalls(stub func(*livekit.GetDataTrackSchemaRequest)) { + fake.handleGetDataTrackSchemaRequestMutex.Lock() + defer fake.handleGetDataTrackSchemaRequestMutex.Unlock() + fake.HandleGetDataTrackSchemaRequestStub = stub +} + +func (fake *FakeLocalParticipant) HandleGetDataTrackSchemaRequestArgsForCall(i int) *livekit.GetDataTrackSchemaRequest { + fake.handleGetDataTrackSchemaRequestMutex.RLock() + defer fake.handleGetDataTrackSchemaRequestMutex.RUnlock() + argsForCall := fake.handleGetDataTrackSchemaRequestArgsForCall[i] + return argsForCall.arg1 +} + func (fake *FakeLocalParticipant) HandleICERestartSDPFragment(arg1 string) (string, error) { fake.handleICERestartSDPFragmentMutex.Lock() ret, specificReturn := fake.handleICERestartSDPFragmentReturnsOnCall[len(fake.handleICERestartSDPFragmentArgsForCall)] @@ -6607,6 +6796,39 @@ func (fake *FakeLocalParticipant) PerformRpcArgsForCall(i int) (*livekit.Perform return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 } +func (fake *FakeLocalParticipant) ProcessGetDataTrackSchemaRequest(arg1 *livekit.GetDataTrackSchemaRequest, arg2 types.Participant) { + fake.processGetDataTrackSchemaRequestMutex.Lock() + fake.processGetDataTrackSchemaRequestArgsForCall = append(fake.processGetDataTrackSchemaRequestArgsForCall, struct { + arg1 *livekit.GetDataTrackSchemaRequest + arg2 types.Participant + }{arg1, arg2}) + stub := fake.ProcessGetDataTrackSchemaRequestStub + fake.recordInvocation("ProcessGetDataTrackSchemaRequest", []interface{}{arg1, arg2}) + fake.processGetDataTrackSchemaRequestMutex.Unlock() + if stub != nil { + fake.ProcessGetDataTrackSchemaRequestStub(arg1, arg2) + } +} + +func (fake *FakeLocalParticipant) ProcessGetDataTrackSchemaRequestCallCount() int { + fake.processGetDataTrackSchemaRequestMutex.RLock() + defer fake.processGetDataTrackSchemaRequestMutex.RUnlock() + return len(fake.processGetDataTrackSchemaRequestArgsForCall) +} + +func (fake *FakeLocalParticipant) ProcessGetDataTrackSchemaRequestCalls(stub func(*livekit.GetDataTrackSchemaRequest, types.Participant)) { + fake.processGetDataTrackSchemaRequestMutex.Lock() + defer fake.processGetDataTrackSchemaRequestMutex.Unlock() + fake.ProcessGetDataTrackSchemaRequestStub = stub +} + +func (fake *FakeLocalParticipant) ProcessGetDataTrackSchemaRequestArgsForCall(i int) (*livekit.GetDataTrackSchemaRequest, types.Participant) { + fake.processGetDataTrackSchemaRequestMutex.RLock() + defer fake.processGetDataTrackSchemaRequestMutex.RUnlock() + argsForCall := fake.processGetDataTrackSchemaRequestArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + func (fake *FakeLocalParticipant) ProtocolVersion() types.ProtocolVersion { fake.protocolVersionMutex.Lock() ret, specificReturn := fake.protocolVersionReturnsOnCall[len(fake.protocolVersionArgsForCall)] diff --git a/pkg/rtc/types/typesfakes/fake_local_participant_listener.go b/pkg/rtc/types/typesfakes/fake_local_participant_listener.go index 4cd0fef83..d585f2d8c 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant_listener.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant_listener.go @@ -42,6 +42,18 @@ type FakeLocalParticipantListener struct { arg1 types.Participant arg2 types.DataTrack } + OnDefineDataTrackSchemaStub func(types.LocalParticipant, *livekit.DataTrackSchemaDefinition) + onDefineDataTrackSchemaMutex sync.RWMutex + onDefineDataTrackSchemaArgsForCall []struct { + arg1 types.LocalParticipant + arg2 *livekit.DataTrackSchemaDefinition + } + OnGetDataTrackSchemaStub func(types.LocalParticipant, *livekit.GetDataTrackSchemaRequest) + onGetDataTrackSchemaMutex sync.RWMutex + onGetDataTrackSchemaArgsForCall []struct { + arg1 types.LocalParticipant + arg2 *livekit.GetDataTrackSchemaRequest + } OnLeaveStub func(types.LocalParticipant, types.ParticipantCloseReason) onLeaveMutex sync.RWMutex onLeaveArgsForCall []struct { @@ -331,6 +343,72 @@ func (fake *FakeLocalParticipantListener) OnDataTrackUnpublishedArgsForCall(i in return argsForCall.arg1, argsForCall.arg2 } +func (fake *FakeLocalParticipantListener) OnDefineDataTrackSchema(arg1 types.LocalParticipant, arg2 *livekit.DataTrackSchemaDefinition) { + fake.onDefineDataTrackSchemaMutex.Lock() + fake.onDefineDataTrackSchemaArgsForCall = append(fake.onDefineDataTrackSchemaArgsForCall, struct { + arg1 types.LocalParticipant + arg2 *livekit.DataTrackSchemaDefinition + }{arg1, arg2}) + stub := fake.OnDefineDataTrackSchemaStub + fake.recordInvocation("OnDefineDataTrackSchema", []interface{}{arg1, arg2}) + fake.onDefineDataTrackSchemaMutex.Unlock() + if stub != nil { + fake.OnDefineDataTrackSchemaStub(arg1, arg2) + } +} + +func (fake *FakeLocalParticipantListener) OnDefineDataTrackSchemaCallCount() int { + fake.onDefineDataTrackSchemaMutex.RLock() + defer fake.onDefineDataTrackSchemaMutex.RUnlock() + return len(fake.onDefineDataTrackSchemaArgsForCall) +} + +func (fake *FakeLocalParticipantListener) OnDefineDataTrackSchemaCalls(stub func(types.LocalParticipant, *livekit.DataTrackSchemaDefinition)) { + fake.onDefineDataTrackSchemaMutex.Lock() + defer fake.onDefineDataTrackSchemaMutex.Unlock() + fake.OnDefineDataTrackSchemaStub = stub +} + +func (fake *FakeLocalParticipantListener) OnDefineDataTrackSchemaArgsForCall(i int) (types.LocalParticipant, *livekit.DataTrackSchemaDefinition) { + fake.onDefineDataTrackSchemaMutex.RLock() + defer fake.onDefineDataTrackSchemaMutex.RUnlock() + argsForCall := fake.onDefineDataTrackSchemaArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeLocalParticipantListener) OnGetDataTrackSchema(arg1 types.LocalParticipant, arg2 *livekit.GetDataTrackSchemaRequest) { + fake.onGetDataTrackSchemaMutex.Lock() + fake.onGetDataTrackSchemaArgsForCall = append(fake.onGetDataTrackSchemaArgsForCall, struct { + arg1 types.LocalParticipant + arg2 *livekit.GetDataTrackSchemaRequest + }{arg1, arg2}) + stub := fake.OnGetDataTrackSchemaStub + fake.recordInvocation("OnGetDataTrackSchema", []interface{}{arg1, arg2}) + fake.onGetDataTrackSchemaMutex.Unlock() + if stub != nil { + fake.OnGetDataTrackSchemaStub(arg1, arg2) + } +} + +func (fake *FakeLocalParticipantListener) OnGetDataTrackSchemaCallCount() int { + fake.onGetDataTrackSchemaMutex.RLock() + defer fake.onGetDataTrackSchemaMutex.RUnlock() + return len(fake.onGetDataTrackSchemaArgsForCall) +} + +func (fake *FakeLocalParticipantListener) OnGetDataTrackSchemaCalls(stub func(types.LocalParticipant, *livekit.GetDataTrackSchemaRequest)) { + fake.onGetDataTrackSchemaMutex.Lock() + defer fake.onGetDataTrackSchemaMutex.Unlock() + fake.OnGetDataTrackSchemaStub = stub +} + +func (fake *FakeLocalParticipantListener) OnGetDataTrackSchemaArgsForCall(i int) (types.LocalParticipant, *livekit.GetDataTrackSchemaRequest) { + fake.onGetDataTrackSchemaMutex.RLock() + defer fake.onGetDataTrackSchemaMutex.RUnlock() + argsForCall := fake.onGetDataTrackSchemaArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + func (fake *FakeLocalParticipantListener) OnLeave(arg1 types.LocalParticipant, arg2 types.ParticipantCloseReason) { fake.onLeaveMutex.Lock() fake.onLeaveArgsForCall = append(fake.onLeaveArgsForCall, struct { diff --git a/pkg/rtc/types/typesfakes/fake_participant.go b/pkg/rtc/types/typesfakes/fake_participant.go index 19498c9c3..5b5d38919 100644 --- a/pkg/rtc/types/typesfakes/fake_participant.go +++ b/pkg/rtc/types/typesfakes/fake_participant.go @@ -13,6 +13,11 @@ import ( ) type FakeParticipant struct { + AddDataTrackSchemaStub func(*livekit.DataTrackSchemaDefinition) + addDataTrackSchemaMutex sync.RWMutex + addDataTrackSchemaArgsForCall []struct { + arg1 *livekit.DataTrackSchemaDefinition + } CanSkipBroadcastStub func() bool canSkipBroadcastMutex sync.RWMutex canSkipBroadcastArgsForCall []struct { @@ -78,6 +83,17 @@ type FakeParticipant struct { result1 float64 result2 bool } + GetDataTrackSchemaStub func(*livekit.DataTrackSchemaId) *livekit.DataTrackSchemaDefinition + getDataTrackSchemaMutex sync.RWMutex + getDataTrackSchemaArgsForCall []struct { + arg1 *livekit.DataTrackSchemaId + } + getDataTrackSchemaReturns struct { + result1 *livekit.DataTrackSchemaDefinition + } + getDataTrackSchemaReturnsOnCall map[int]struct { + result1 *livekit.DataTrackSchemaDefinition + } GetLoggerStub func() logger.Logger getLoggerMutex sync.RWMutex getLoggerArgsForCall []struct { @@ -361,6 +377,38 @@ type FakeParticipant struct { invocationsMutex sync.RWMutex } +func (fake *FakeParticipant) AddDataTrackSchema(arg1 *livekit.DataTrackSchemaDefinition) { + fake.addDataTrackSchemaMutex.Lock() + fake.addDataTrackSchemaArgsForCall = append(fake.addDataTrackSchemaArgsForCall, struct { + arg1 *livekit.DataTrackSchemaDefinition + }{arg1}) + stub := fake.AddDataTrackSchemaStub + fake.recordInvocation("AddDataTrackSchema", []interface{}{arg1}) + fake.addDataTrackSchemaMutex.Unlock() + if stub != nil { + fake.AddDataTrackSchemaStub(arg1) + } +} + +func (fake *FakeParticipant) AddDataTrackSchemaCallCount() int { + fake.addDataTrackSchemaMutex.RLock() + defer fake.addDataTrackSchemaMutex.RUnlock() + return len(fake.addDataTrackSchemaArgsForCall) +} + +func (fake *FakeParticipant) AddDataTrackSchemaCalls(stub func(*livekit.DataTrackSchemaDefinition)) { + fake.addDataTrackSchemaMutex.Lock() + defer fake.addDataTrackSchemaMutex.Unlock() + fake.AddDataTrackSchemaStub = stub +} + +func (fake *FakeParticipant) AddDataTrackSchemaArgsForCall(i int) *livekit.DataTrackSchemaDefinition { + fake.addDataTrackSchemaMutex.RLock() + defer fake.addDataTrackSchemaMutex.RUnlock() + argsForCall := fake.addDataTrackSchemaArgsForCall[i] + return argsForCall.arg1 +} + func (fake *FakeParticipant) CanSkipBroadcast() bool { fake.canSkipBroadcastMutex.Lock() ret, specificReturn := fake.canSkipBroadcastReturnsOnCall[len(fake.canSkipBroadcastArgsForCall)] @@ -692,6 +740,67 @@ func (fake *FakeParticipant) GetAudioLevelReturnsOnCall(i int, result1 float64, }{result1, result2} } +func (fake *FakeParticipant) GetDataTrackSchema(arg1 *livekit.DataTrackSchemaId) *livekit.DataTrackSchemaDefinition { + fake.getDataTrackSchemaMutex.Lock() + ret, specificReturn := fake.getDataTrackSchemaReturnsOnCall[len(fake.getDataTrackSchemaArgsForCall)] + fake.getDataTrackSchemaArgsForCall = append(fake.getDataTrackSchemaArgsForCall, struct { + arg1 *livekit.DataTrackSchemaId + }{arg1}) + stub := fake.GetDataTrackSchemaStub + fakeReturns := fake.getDataTrackSchemaReturns + fake.recordInvocation("GetDataTrackSchema", []interface{}{arg1}) + fake.getDataTrackSchemaMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeParticipant) GetDataTrackSchemaCallCount() int { + fake.getDataTrackSchemaMutex.RLock() + defer fake.getDataTrackSchemaMutex.RUnlock() + return len(fake.getDataTrackSchemaArgsForCall) +} + +func (fake *FakeParticipant) GetDataTrackSchemaCalls(stub func(*livekit.DataTrackSchemaId) *livekit.DataTrackSchemaDefinition) { + fake.getDataTrackSchemaMutex.Lock() + defer fake.getDataTrackSchemaMutex.Unlock() + fake.GetDataTrackSchemaStub = stub +} + +func (fake *FakeParticipant) GetDataTrackSchemaArgsForCall(i int) *livekit.DataTrackSchemaId { + fake.getDataTrackSchemaMutex.RLock() + defer fake.getDataTrackSchemaMutex.RUnlock() + argsForCall := fake.getDataTrackSchemaArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeParticipant) GetDataTrackSchemaReturns(result1 *livekit.DataTrackSchemaDefinition) { + fake.getDataTrackSchemaMutex.Lock() + defer fake.getDataTrackSchemaMutex.Unlock() + fake.GetDataTrackSchemaStub = nil + fake.getDataTrackSchemaReturns = struct { + result1 *livekit.DataTrackSchemaDefinition + }{result1} +} + +func (fake *FakeParticipant) GetDataTrackSchemaReturnsOnCall(i int, result1 *livekit.DataTrackSchemaDefinition) { + fake.getDataTrackSchemaMutex.Lock() + defer fake.getDataTrackSchemaMutex.Unlock() + fake.GetDataTrackSchemaStub = nil + if fake.getDataTrackSchemaReturnsOnCall == nil { + fake.getDataTrackSchemaReturnsOnCall = make(map[int]struct { + result1 *livekit.DataTrackSchemaDefinition + }) + } + fake.getDataTrackSchemaReturnsOnCall[i] = struct { + result1 *livekit.DataTrackSchemaDefinition + }{result1} +} + func (fake *FakeParticipant) GetLogger() logger.Logger { fake.getLoggerMutex.Lock() ret, specificReturn := fake.getLoggerReturnsOnCall[len(fake.getLoggerArgsForCall)] diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 65eeb5eda..b70844bd0 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -520,8 +520,9 @@ func (r *RoomManager) StartSession( DatachannelLossyTargetLatency: r.config.RTC.DatachannelLossyTargetLatency, FireOnTrackBySdp: true, UseSinglePeerConnection: pi.UseSinglePeerConnection, - EnableDataTracks: r.config.EnableDataTracks, - EnableRTPStreamRestartDetection: r.config.RTC.EnableRTPStreamRestartDetection, + EnableDataTracks: r.config.EnableDataTracks, + EnableParticipantAsyncAttributes: r.config.EnableParticipantAsyncAttributes, + EnableRTPStreamRestartDetection: r.config.RTC.EnableRTPStreamRestartDetection, }) if err != nil { return err diff --git a/test/integration_helpers.go b/test/integration_helpers.go index 3ccde58d9..367294e46 100644 --- a/test/integration_helpers.go +++ b/test/integration_helpers.go @@ -80,9 +80,13 @@ func setupSingleNodeTest(name string) (*service.LivekitServer, func()) { } func setupMultiNodeTest(name string) (*service.LivekitServer, *service.LivekitServer, func()) { + return setupMultiNodeTestWithConfig(name, nil) +} + +func setupMultiNodeTestWithConfig(name string, configUpdater func(*config.Config)) (*service.LivekitServer, *service.LivekitServer, func()) { logger.Infow("----------------STARTING TEST----------------", "test", name) - s1 := createMultiNodeServer(guid.New(nodeID1), defaultServerPort) - s2 := createMultiNodeServer(guid.New(nodeID2), secondServerPort) + s1 := createMultiNodeServer(guid.New(nodeID1), defaultServerPort, configUpdater) + s2 := createMultiNodeServer(guid.New(nodeID2), secondServerPort, configUpdater) go s1.Start() go s2.Start() @@ -190,7 +194,7 @@ func createSingleNodeServer(configUpdater func(*config.Config)) *service.Livekit return s } -func createMultiNodeServer(nodeID string, port uint32) *service.LivekitServer { +func createMultiNodeServer(nodeID string, port uint32, configUpdater func(*config.Config)) *service.LivekitServer { var err error conf, err := config.NewConfig("", true, nil, nil) if err != nil { @@ -202,6 +206,9 @@ func createMultiNodeServer(nodeID string, port uint32) *service.LivekitServer { conf.Redis.Address = "localhost:6379" conf.Keys = map[string]string{testApiKey: testApiSecret} conf.EnableDataTracks = true + if configUpdater != nil { + configUpdater(conf) + } currentNode, err := routing.NewLocalNode(conf) if err != nil { diff --git a/test/multinode_test.go b/test/multinode_test.go index 9b4d5acea..c26f9d468 100644 --- a/test/multinode_test.go +++ b/test/multinode_test.go @@ -24,6 +24,7 @@ import ( "github.com/livekit/protocol/auth" "github.com/livekit/protocol/livekit" + "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/rtc" "github.com/livekit/livekit-server/pkg/testutils" "github.com/livekit/livekit-server/test/client" @@ -425,3 +426,113 @@ func TestCloseDisconnectedParticipantOnSignalClose(t *testing.T) { }) } } + +func TestMultiNodeAsyncAttributes(t *testing.T) { + if testing.Short() { + t.SkipNow() + return + } + + _, _, finish := setupMultiNodeTestWithConfig("TestMultiNodeAsyncAttributes", func(c *config.Config) { + c.EnableParticipantAsyncAttributes = true + c.Limit.MaxAsyncAttributesSize = 1024 + }) + defer finish() + + for _, testRTCServicePath := range testRTCServicePaths { + t.Run(fmt.Sprintf("testRTCServicePath=%s", testRTCServicePath.String()), func(t *testing.T) { + pubCapture := &asyncAttributesCapture{} + subCapture := &asyncAttributesCapture{} + + // publisher on node 1, subscriber on node 2 + pub := createRTCClient("pub", defaultServerPort, testRTCServicePath, &client.Options{ + AutoSubscribe: true, + SignalResponseInterceptor: pubCapture.interceptor(), + }) + sub := createRTCClient("sub", secondServerPort, testRTCServicePath, &client.Options{ + AutoSubscribe: true, + SignalResponseInterceptor: subCapture.interceptor(), + }) + waitUntilConnected(t, pub, sub) + defer stopClients(pub, sub) + + // wait for both nodes to see each other so the get request routes correctly + testutils.WithTimeout(t, func() string { + if sub.GetRemoteParticipant(pub.ID()) == nil { + return "sub does not see pub yet" + } + return "" + }) + + schemaID := &livekit.DataTrackSchemaId{ + Name: "schema-multinode", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + } + definition := []byte("multinode-definition") + + require.NoError(t, pub.SendRequest(&livekit.SignalRequest{ + Message: &livekit.SignalRequest_DefineDataTrackSchema{ + DefineDataTrackSchema: &livekit.DefineDataTrackSchemaRequest{ + SchemaDefinition: &livekit.DataTrackSchemaDefinition{ + Id: schemaID, + Definition: definition, + }, + }, + }, + })) + + // give the publisher's node time to apply the definition + time.Sleep(syncDelay) + require.Equal(t, 0, pubCapture.requestResponseCount(), "publisher should not receive an error response on success") + + // subscriber on a different node asks for the schema; the request routes + // across nodes to the publisher. + require.NoError(t, sub.SendRequest(&livekit.SignalRequest{ + Message: &livekit.SignalRequest_GetDataTrackSchema{ + GetDataTrackSchema: &livekit.GetDataTrackSchemaRequest{ + ParticipantIdentity: "pub", + SchemaId: schemaID, + }, + }, + })) + + testutils.WithTimeout(t, func() string { + resp := subCapture.takeSchemaResponse() + if resp == nil { + return "subscriber did not receive schema response" + } + if resp.SchemaDefinition == nil { + return "schema response missing definition" + } + if resp.SchemaDefinition.Id.Name != schemaID.Name { + return fmt.Sprintf("expected schema name %s, got %s", schemaID.Name, resp.SchemaDefinition.Id.Name) + } + if string(resp.SchemaDefinition.Definition) != string(definition) { + return fmt.Sprintf("expected definition %q, got %q", definition, resp.SchemaDefinition.Definition) + } + return "" + }) + + // requesting an unknown publisher identity should return NOT_FOUND + require.NoError(t, sub.SendRequest(&livekit.SignalRequest{ + Message: &livekit.SignalRequest_GetDataTrackSchema{ + GetDataTrackSchema: &livekit.GetDataTrackSchemaRequest{ + ParticipantIdentity: "unknown-publisher", + SchemaId: schemaID, + }, + }, + })) + + testutils.WithTimeout(t, func() string { + rr := subCapture.takeRequestResponse() + if rr == nil { + return "subscriber did not receive RequestResponse for unknown publisher" + } + if rr.Reason != livekit.RequestResponse_NOT_FOUND { + return fmt.Sprintf("expected NOT_FOUND, got %s", rr.Reason) + } + return "" + }) + }) + } +} diff --git a/test/singlenode_test.go b/test/singlenode_test.go index 6d571dadf..bf51a04da 100644 --- a/test/singlenode_test.go +++ b/test/singlenode_test.go @@ -1512,3 +1512,269 @@ func TestTurnAuthFailure(t *testing.T) { }) } } + +// asyncAttributesCapture buffers RequestResponse and GetDataTrackSchemaResponse messages +// sent to a test client so they can be asserted on. Other messages flow through to the +// default handler. +type asyncAttributesCapture struct { + mu sync.Mutex + requestResponses []*livekit.RequestResponse + schemaResponses []*livekit.GetDataTrackSchemaResponse +} + +func (c *asyncAttributesCapture) interceptor() testclient.SignalResponseInterceptor { + return func(msg *livekit.SignalResponse, next testclient.SignalResponseHandler) error { + switch m := msg.Message.(type) { + case *livekit.SignalResponse_RequestResponse: + c.mu.Lock() + c.requestResponses = append(c.requestResponses, m.RequestResponse) + c.mu.Unlock() + case *livekit.SignalResponse_GetDataTrackSchemaResponse: + c.mu.Lock() + c.schemaResponses = append(c.schemaResponses, m.GetDataTrackSchemaResponse) + c.mu.Unlock() + } + return next(msg) + } +} + +func (c *asyncAttributesCapture) takeRequestResponse() *livekit.RequestResponse { + c.mu.Lock() + defer c.mu.Unlock() + if len(c.requestResponses) == 0 { + return nil + } + rr := c.requestResponses[0] + c.requestResponses = c.requestResponses[1:] + return rr +} + +func (c *asyncAttributesCapture) takeSchemaResponse() *livekit.GetDataTrackSchemaResponse { + c.mu.Lock() + defer c.mu.Unlock() + if len(c.schemaResponses) == 0 { + return nil + } + sr := c.schemaResponses[0] + c.schemaResponses = c.schemaResponses[1:] + return sr +} + +func (c *asyncAttributesCapture) requestResponseCount() int { + c.mu.Lock() + defer c.mu.Unlock() + return len(c.requestResponses) +} + +func setupAsyncAttributesServer(t *testing.T, name string, enable bool) (*service.LivekitServer, func()) { + logger.Infow("----------------STARTING TEST----------------", "test", name) + s := createSingleNodeServer(func(c *config.Config) { + c.EnableParticipantAsyncAttributes = enable + c.Limit.MaxAsyncAttributesSize = 1024 + }) + go func() { + if err := s.Start(); err != nil { + logger.Errorw("server returned error", err) + } + }() + waitForServerToStart(s) + return s, func() { + s.Stop(true) + logger.Infow("----------------FINISHING TEST----------------", "test", name) + } +} + +func TestSingleNodeAsyncAttributes(t *testing.T) { + if testing.Short() { + t.SkipNow() + return + } + + _, finish := setupAsyncAttributesServer(t, "TestSingleNodeAsyncAttributes", true) + defer finish() + + for _, testRTCServicePath := range testRTCServicePaths { + t.Run(fmt.Sprintf("testRTCServicePath=%s", testRTCServicePath.String()), func(t *testing.T) { + pubCapture := &asyncAttributesCapture{} + subCapture := &asyncAttributesCapture{} + + pub := createRTCClient("pub", defaultServerPort, testRTCServicePath, &testclient.Options{ + AutoSubscribe: true, + SignalResponseInterceptor: pubCapture.interceptor(), + }) + sub := createRTCClient("sub", defaultServerPort, testRTCServicePath, &testclient.Options{ + AutoSubscribe: true, + SignalResponseInterceptor: subCapture.interceptor(), + }) + waitUntilConnected(t, pub, sub) + defer stopClients(pub, sub) + + schemaID := &livekit.DataTrackSchemaId{ + Name: "schema-1", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + } + definition := []byte("definition-bytes") + + // publisher defines a schema + require.NoError(t, pub.SendRequest(&livekit.SignalRequest{ + Message: &livekit.SignalRequest_DefineDataTrackSchema{ + DefineDataTrackSchema: &livekit.DefineDataTrackSchemaRequest{ + SchemaDefinition: &livekit.DataTrackSchemaDefinition{ + Id: schemaID, + Definition: definition, + }, + }, + }, + })) + + // give the server a moment to process; success path sends no response + time.Sleep(syncDelay) + require.Equal(t, 0, pubCapture.requestResponseCount(), "publisher should not receive an error response on success") + + // subscriber asks for the schema + require.NoError(t, sub.SendRequest(&livekit.SignalRequest{ + Message: &livekit.SignalRequest_GetDataTrackSchema{ + GetDataTrackSchema: &livekit.GetDataTrackSchemaRequest{ + ParticipantIdentity: "pub", + SchemaId: schemaID, + }, + }, + })) + + testutils.WithTimeout(t, func() string { + resp := subCapture.takeSchemaResponse() + if resp == nil { + return "subscriber did not receive schema response" + } + if resp.SchemaDefinition == nil { + return "schema response missing definition" + } + if resp.SchemaDefinition.Id.Name != schemaID.Name { + return fmt.Sprintf("expected schema name %s, got %s", schemaID.Name, resp.SchemaDefinition.Id.Name) + } + if string(resp.SchemaDefinition.Definition) != string(definition) { + return fmt.Sprintf("expected definition %q, got %q", definition, resp.SchemaDefinition.Definition) + } + return "" + }) + + // subscriber asks for an unknown schema on a known publisher + require.NoError(t, sub.SendRequest(&livekit.SignalRequest{ + Message: &livekit.SignalRequest_GetDataTrackSchema{ + GetDataTrackSchema: &livekit.GetDataTrackSchemaRequest{ + ParticipantIdentity: "pub", + SchemaId: &livekit.DataTrackSchemaId{ + Name: "does-not-exist", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + }, + }, + }, + })) + + testutils.WithTimeout(t, func() string { + rr := subCapture.takeRequestResponse() + if rr == nil { + return "subscriber did not receive RequestResponse for missing schema" + } + if rr.Reason != livekit.RequestResponse_NOT_FOUND { + return fmt.Sprintf("expected NOT_FOUND, got %s", rr.Reason) + } + return "" + }) + + // subscriber asks for a schema on an unknown publisher identity + require.NoError(t, sub.SendRequest(&livekit.SignalRequest{ + Message: &livekit.SignalRequest_GetDataTrackSchema{ + GetDataTrackSchema: &livekit.GetDataTrackSchemaRequest{ + ParticipantIdentity: "unknown-publisher", + SchemaId: schemaID, + }, + }, + })) + + testutils.WithTimeout(t, func() string { + rr := subCapture.takeRequestResponse() + if rr == nil { + return "subscriber did not receive RequestResponse for unknown publisher" + } + if rr.Reason != livekit.RequestResponse_NOT_FOUND { + return fmt.Sprintf("expected NOT_FOUND, got %s", rr.Reason) + } + return "" + }) + + // publisher sends an invalid define (empty id name) + require.NoError(t, pub.SendRequest(&livekit.SignalRequest{ + Message: &livekit.SignalRequest_DefineDataTrackSchema{ + DefineDataTrackSchema: &livekit.DefineDataTrackSchemaRequest{ + SchemaDefinition: &livekit.DataTrackSchemaDefinition{ + Id: &livekit.DataTrackSchemaId{ + Name: "", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + }, + Definition: definition, + }, + }, + }, + })) + + testutils.WithTimeout(t, func() string { + rr := pubCapture.takeRequestResponse() + if rr == nil { + return "publisher did not receive RequestResponse for invalid define" + } + if rr.Reason != livekit.RequestResponse_INVALID_REQUEST { + return fmt.Sprintf("expected INVALID_REQUEST, got %s", rr.Reason) + } + return "" + }) + }) + } +} + +func TestSingleNodeAsyncAttributesDisabled(t *testing.T) { + if testing.Short() { + t.SkipNow() + return + } + + _, finish := setupAsyncAttributesServer(t, "TestSingleNodeAsyncAttributesDisabled", false) + defer finish() + + for _, testRTCServicePath := range testRTCServicePaths { + t.Run(fmt.Sprintf("testRTCServicePath=%s", testRTCServicePath.String()), func(t *testing.T) { + pubCapture := &asyncAttributesCapture{} + pub := createRTCClient("pub", defaultServerPort, testRTCServicePath, &testclient.Options{ + AutoSubscribe: true, + SignalResponseInterceptor: pubCapture.interceptor(), + }) + waitUntilConnected(t, pub) + defer stopClients(pub) + + require.NoError(t, pub.SendRequest(&livekit.SignalRequest{ + Message: &livekit.SignalRequest_DefineDataTrackSchema{ + DefineDataTrackSchema: &livekit.DefineDataTrackSchemaRequest{ + SchemaDefinition: &livekit.DataTrackSchemaDefinition{ + Id: &livekit.DataTrackSchemaId{ + Name: "schema-1", + Encoding: livekit.DataTrackSchemaEncoding_DATA_TRACK_SCHEMA_ENCODING_PROTOBUF, + }, + Definition: []byte("definition-bytes"), + }, + }, + }, + })) + + testutils.WithTimeout(t, func() string { + rr := pubCapture.takeRequestResponse() + if rr == nil { + return "publisher did not receive RequestResponse" + } + if rr.Reason != livekit.RequestResponse_NOT_ALLOWED { + return fmt.Sprintf("expected NOT_ALLOWED, got %s", rr.Reason) + } + return "" + }) + }) + } +}