diff --git a/go.mod b/go.mod index 938a14b22..89a90de11 100644 --- a/go.mod +++ b/go.mod @@ -21,7 +21,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 github.com/livekit/mediatransportutil v0.0.0-20260608063931-a3417d38cda0 - github.com/livekit/protocol v1.48.1-0.20260622160555-777bf63c9d52 + github.com/livekit/protocol v1.48.1-0.20260623210753-2e1bfd81dd63 github.com/livekit/psrpc v0.7.2 github.com/mackerelio/go-osstat v0.2.7 github.com/magefile/mage v1.17.2 @@ -149,7 +149,7 @@ require ( golang.org/x/sys v0.46.0 // indirect golang.org/x/text v0.38.0 // indirect golang.org/x/tools v0.46.0 // indirect - google.golang.org/genproto/googleapis/api v0.0.0-20260618152121-87f3d3e198d3 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20260618152121-87f3d3e198d3 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260622175928-b703f567277d // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260622175928-b703f567277d // indirect google.golang.org/grpc v1.81.1 // indirect ) diff --git a/go.sum b/go.sum index 37cbc525f..29ca2a832 100644 --- a/go.sum +++ b/go.sum @@ -160,8 +160,8 @@ 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-20260608063931-a3417d38cda0 h1:XHNNzebIKZRkLimla/hFGrAIX5EMWHctrgt3hLw7s+I= github.com/livekit/mediatransportutil v0.0.0-20260608063931-a3417d38cda0/go.mod h1:o8CFmAdrVwzJNOCsQCLUzXRjokkufNshnQHOe4fRaqU= -github.com/livekit/protocol v1.48.1-0.20260622160555-777bf63c9d52 h1:9obg71kYBNSegrEAkcqBebi9KxxFDyS7GXXcUkXAQVM= -github.com/livekit/protocol v1.48.1-0.20260622160555-777bf63c9d52/go.mod h1:jO+y05AU9Ec4JswDyuzKCZ4bhziOS0CzMqgnbj60Dzs= +github.com/livekit/protocol v1.48.1-0.20260623210753-2e1bfd81dd63 h1:Rj9/54oztXeioAwUkukXBwcPE/GxL97WylFd1m7V2pQ= +github.com/livekit/protocol v1.48.1-0.20260623210753-2e1bfd81dd63/go.mod h1:jO+y05AU9Ec4JswDyuzKCZ4bhziOS0CzMqgnbj60Dzs= github.com/livekit/psrpc v0.7.2 h1:6oZ+NODJ2pLyaT6VqDq1F4Qc/3TpDUSpyphj/P9MhQc= github.com/livekit/psrpc v0.7.2/go.mod h1:rAI+m2+/cb4x9RXhLRtUx5ZwdfjjXOl4zi46IjEetaw= github.com/mackerelio/go-osstat v0.2.7 h1:TCavZi10wF49bT6iQZ9eT2keGZQpC69MTDfdJej5e94= @@ -419,10 +419,10 @@ golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8T golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= -google.golang.org/genproto/googleapis/api v0.0.0-20260618152121-87f3d3e198d3 h1:ctPmKL12ZsoKAlmPUsoW70zEDiYF+/H6aLieXxgAU0k= -google.golang.org/genproto/googleapis/api v0.0.0-20260618152121-87f3d3e198d3/go.mod h1:Z4WJ5pJOYWFWcHEQUelD5QaZDknIQkpIL/+fyJOT9+A= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260618152121-87f3d3e198d3 h1:phvBWCAQMGN1945mp5fjCXP6jEF0+a0+4TjokS4sxNY= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260618152121-87f3d3e198d3/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/genproto/googleapis/api v0.0.0-20260622175928-b703f567277d h1:xr2lwHI91bn3UiXcnyzRMQjp2LRiM8wEHzwUaE0YhTs= +google.golang.org/genproto/googleapis/api v0.0.0-20260622175928-b703f567277d/go.mod h1:O0ZOWSrfWfJ+Z5HbwZ+wNtHsg/vk1k2C/w67eww8PfQ= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260622175928-b703f567277d h1:mpAgMyM9vQHxycBlDq50y1VHpfSfVwzXvrQKtYbXuUY= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260622175928-b703f567277d/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= google.golang.org/grpc v1.81.1 h1:VnnIIZ88UzOOKLukQi+ImGz8O1Wdp8nAGGnvOfEIWQQ= google.golang.org/grpc v1.81.1/go.mod h1:xGH9GfzOyMTGIOXBJmXt+BX/V0kcdQbdcuwQ/zNw42I= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= diff --git a/pkg/config/config.go b/pkg/config/config.go index 4393f7f80..f85f6a841 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -91,6 +91,8 @@ type Config struct { EnableDataTracks bool `yaml:"enable_data_tracks,omitempty"` + EnableParticipantDataBlob bool `yaml:"enable_participant_data_blob,omitempty"` + API APIConfig `yaml:"api,omitempty"` } @@ -280,6 +282,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"` @@ -291,6 +295,9 @@ 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"` + + MaxDataBlobKeyLength int `yaml:"max_data_blob_key_length,omitempty"` + MaxDataBlobSize uint32 `yaml:"max_data_blobs_size,omitempty"` } func (l LimitConfig) CheckRoomNameLength(name string) bool { @@ -321,6 +328,36 @@ func (l LimitConfig) CheckAttributesSize(attributes map[string]string) bool { return uint32(total) <= l.MaxAttributesSize } +func (l LimitConfig) CheckDataBlobKeyLength(key string) bool { + return l.MaxDataBlobKeyLength == 0 || len(key) <= l.MaxDataBlobKeyLength +} + +func (l LimitConfig) CheckDataBlobsSize(dataBlobs []*livekit.DataBlob) bool { + if l.MaxDataBlobSize == 0 { + return true + } + + total := 0 + for _, dataBlob := range dataBlobs { + total += len(dataBlob.GetKey().String()) + len(dataBlob.Contents) + } + return uint32(total) <= l.MaxDataBlobSize +} + +func (l LimitConfig) CanAddDataBlob(dataBlobs []*livekit.DataBlob, toAdd *livekit.DataBlob) bool { + if l.MaxDataBlobSize == 0 { + return true + } + + total := 0 + for _, dataBlob := range dataBlobs { + total += len(dataBlob.Key.String()) + len(dataBlob.Contents) + } + return uint32(total+len(toAdd.GetKey().String())+len(toAdd.Contents)) <= l.MaxDataBlobSize +} + +// --------------------------------- + type IngressConfig struct { RTMPBaseURL string `yaml:"rtmp_base_url,omitempty"` WHIPBaseURL string `yaml:"whip_base_url,omitempty"` @@ -443,6 +480,8 @@ var DefaultConfig = Config{ MaxRoomNameLength: 256, MaxParticipantIdentityLength: 256, MaxParticipantNameLength: 256, + MaxDataBlobKeyLength: 256, + MaxDataBlobSize: 64000, }, Logging: LoggingConfig{ PionLevel: "error", diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index f510cf5af..1cf5bb5b2 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -225,6 +225,7 @@ type ParticipantParams struct { EnableRTPStreamRestartDetection bool ForceBackupCodecPolicySimulcast bool DisableTransceiverReuseForE2EE bool + EnableParticipantDataBlob bool EnableStartAtDesiredQuality bool } @@ -334,6 +335,8 @@ type ParticipantImpl struct { rpcLock sync.Mutex rpcPendingAcks map[string]*utils.DataChannelRpcPendingAckHandler rpcPendingResponses map[string]*utils.DataChannelRpcPendingResponseHandler + + dataBlob *ParticipantDataBlob } func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { @@ -372,6 +375,9 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { telemetryGuard: &telemetry.ReferenceGuard{}, nextSubscribedDataTrackHandle: uint16(rand.Intn(256)), requireBroadcast: params.Grants.Metadata != "" || len(params.Grants.Attributes) != 0, + dataBlob: NewParticipantDataBlob(ParticipantDataBlobParams{ + Logger: params.Logger, + }), } p.setupSignalling() diff --git a/pkg/rtc/participant_data_blob.go b/pkg/rtc/participant_data_blob.go new file mode 100644 index 000000000..74512d569 --- /dev/null +++ b/pkg/rtc/participant_data_blob.go @@ -0,0 +1,90 @@ +// 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" + + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" +) + +type ParticipantDataBlobParams struct { + Logger logger.Logger +} + +type ParticipantDataBlob struct { + params ParticipantDataBlobParams + lock sync.Mutex + blobs map[string]*livekit.DataBlob +} + +func NewParticipantDataBlob(params ParticipantDataBlobParams) *ParticipantDataBlob { + return &ParticipantDataBlob{ + params: params, + blobs: make(map[string]*livekit.DataBlob), + } +} + +func (p *ParticipantDataBlob) Add(db *livekit.DataBlob) { + p.lock.Lock() + defer p.lock.Unlock() + + if db.Key == nil { + return + } + + p.blobs[db.Key.String()] = db +} + +func (p *ParticipantDataBlob) Delete(dbKey *livekit.DataBlobKey) { + p.lock.Lock() + defer p.lock.Unlock() + + if dbKey == nil { + return + } + + delete(p.blobs, dbKey.String()) +} + +func (p *ParticipantDataBlob) Get(dbKey *livekit.DataBlobKey) *livekit.DataBlob { + p.lock.Lock() + defer p.lock.Unlock() + + if dbKey == nil { + return nil + } + + db, ok := p.blobs[dbKey.String()] + if !ok { + return nil + } + + return db +} + +func (p *ParticipantDataBlob) GetAll() []*livekit.DataBlob { + p.lock.Lock() + defer p.lock.Unlock() + + all := make([]*livekit.DataBlob, 0, len(p.blobs)) + for _, db := range p.blobs { + all = append(all, db) + } + return all +} + +// ------------------------------- diff --git a/pkg/rtc/participant_data_blob_handler.go b/pkg/rtc/participant_data_blob_handler.go new file mode 100644 index 000000000..e6ebcc8ed --- /dev/null +++ b/pkg/rtc/participant_data_blob_handler.go @@ -0,0 +1,113 @@ +// 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" +) + +func (p *ParticipantImpl) HandleStoreDataBlobRequest(req *livekit.StoreDataBlobRequest) { + if !p.params.EnableParticipantDataBlob { + p.pubLogger.Warnw("data blob not enabled", nil, "req", logger.Proto(req)) + p.sendRequestResponse(&livekit.RequestResponse{ + RequestId: req.RequestId, + Reason: livekit.RequestResponse_NOT_ALLOWED, + Message: "data blob not enabled", + }) + return + } + + if req.Blob == nil || req.Blob.Key == nil || len(req.Blob.Key.String()) == 0 || !p.params.LimitConfig.CheckDataBlobKeyLength(req.Blob.Key.String()) { + p.pubLogger.Warnw("data blob is invalid", nil, "req", logger.Proto(req)) + p.sendRequestResponse(&livekit.RequestResponse{ + RequestId: req.RequestId, + Reason: livekit.RequestResponse_INVALID_REQUEST, + Message: "data blob is invalid", + }) + return + } + + if len(req.Blob.Contents) == 0 { + p.sendRequestResponse(&livekit.RequestResponse{ + RequestId: req.RequestId, + Reason: livekit.RequestResponse_INVALID_REQUEST, + Message: "data blob is empty", + }) + return + } + + if !p.params.LimitConfig.CanAddDataBlob(p.dataBlob.GetAll(), req.Blob) { + p.sendRequestResponse(&livekit.RequestResponse{ + RequestId: req.RequestId, + Reason: livekit.RequestResponse_LIMIT_EXCEEDED, + Message: "async attribute definition exceeds limit", + }) + return + } + + p.AddDataBlob(req.Blob) + p.listener().OnStoreDataBlob(p, req.Blob) + p.sendStoreDataBlobResponse(req.RequestId, req.Blob.Key) +} + +func (p *ParticipantImpl) HandleGetDataBlobRequest(req *livekit.GetDataBlobRequest) { + if req.Key == nil { + p.sendRequestResponse(&livekit.RequestResponse{ + RequestId: req.RequestId, + Reason: livekit.RequestResponse_INVALID_REQUEST, + Message: "data blob key is required", + }) + return + } + + p.listener().OnGetDataBlob(p, req) +} + +func (p *ParticipantImpl) AddDataBlob(dataBlob *livekit.DataBlob) { + p.dataBlob.Add(dataBlob) +} + +func (p *ParticipantImpl) GetDataBlob(key *livekit.DataBlobKey) *livekit.DataBlob { + return p.dataBlob.Get(key) +} + +func (p *ParticipantImpl) ProcessGetDataBlobRequest(req *livekit.GetDataBlobRequest, publisher types.Participant) { + if publisher == nil { + p.sendRequestResponse(&livekit.RequestResponse{ + RequestId: req.RequestId, + Reason: livekit.RequestResponse_NOT_FOUND, + Message: "participant not found", + }) + return + } + + dataBlob := publisher.GetDataBlob(req.Key) + if dataBlob == nil { + p.sendRequestResponse(&livekit.RequestResponse{ + RequestId: req.RequestId, + Reason: livekit.RequestResponse_NOT_FOUND, + Message: "data blob not found", + }) + return + } + + p.sendGetDataBlobResponse(req.RequestId, dataBlob) +} + +func (p *ParticipantImpl) GetAllDataBlob() []*livekit.DataBlob { + return p.dataBlob.GetAll() +} diff --git a/pkg/rtc/participant_data_blob_handler_test.go b/pkg/rtc/participant_data_blob_handler_test.go new file mode 100644 index 000000000..411a6a203 --- /dev/null +++ b/pkg/rtc/participant_data_blob_handler_test.go @@ -0,0 +1,300 @@ +// 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 newParticipantWithDataBlob(t *testing.T, enabled bool, maxKeyLength int, maxSize uint32) *ParticipantImpl { + t.Helper() + p := newParticipantForTest("test") + p.params.EnableParticipantDataBlob = enabled + p.params.LimitConfig = config.LimitConfig{ + MaxDataBlobKeyLength: maxKeyLength, + MaxDataBlobSize: 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 TestHandleStoreDataBlobRequest(t *testing.T) { + t.Run("returns NOT_ALLOWED when feature not enabled", func(t *testing.T) { + p := newParticipantWithDataBlob(t, false, 0, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + req := &livekit.StoreDataBlobRequest{ + Blob: &livekit.DataBlob{ + Key: genericKey("blob-1"), + Contents: []byte("def"), + }, + } + p.HandleStoreDataBlobRequest(req) + + require.Equal(t, 1, sink.WriteMessageCallCount()) + rr := lastRequestResponse(t, sink, 0) + require.Equal(t, livekit.RequestResponse_NOT_ALLOWED, rr.Reason) + require.Empty(t, p.dataBlob.GetAll()) + }) + + t.Run("returns INVALID_REQUEST when blob is nil", func(t *testing.T) { + p := newParticipantWithDataBlob(t, true, 0, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + p.HandleStoreDataBlobRequest(&livekit.StoreDataBlobRequest{}) + + require.Equal(t, 1, sink.WriteMessageCallCount()) + rr := lastRequestResponse(t, sink, 0) + require.Equal(t, livekit.RequestResponse_INVALID_REQUEST, rr.Reason) + require.Empty(t, p.dataBlob.GetAll()) + }) + + t.Run("returns INVALID_REQUEST when key is nil", func(t *testing.T) { + p := newParticipantWithDataBlob(t, true, 0, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + p.HandleStoreDataBlobRequest(&livekit.StoreDataBlobRequest{ + Blob: &livekit.DataBlob{ + Contents: []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 key has no oneof set", func(t *testing.T) { + p := newParticipantWithDataBlob(t, true, 0, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + p.HandleStoreDataBlobRequest(&livekit.StoreDataBlobRequest{ + Blob: &livekit.DataBlob{ + Key: &livekit.DataBlobKey{}, + Contents: []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 key exceeds length limit", func(t *testing.T) { + p := newParticipantWithDataBlob(t, true, 5, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + p.HandleStoreDataBlobRequest(&livekit.StoreDataBlobRequest{ + Blob: &livekit.DataBlob{ + Key: genericKey(strings.Repeat("a", 64)), + Contents: []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 contents is empty", func(t *testing.T) { + p := newParticipantWithDataBlob(t, true, 0, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + p.HandleStoreDataBlobRequest(&livekit.StoreDataBlobRequest{ + Blob: &livekit.DataBlob{ + Key: genericKey("blob-1"), + }, + }) + + require.Equal(t, 1, sink.WriteMessageCallCount()) + rr := lastRequestResponse(t, sink, 0) + require.Equal(t, livekit.RequestResponse_INVALID_REQUEST, rr.Reason) + require.Empty(t, p.dataBlob.GetAll()) + }) + + t.Run("returns LIMIT_EXCEEDED when adding would breach the limit", func(t *testing.T) { + p := newParticipantWithDataBlob(t, true, 0, 16) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + p.HandleStoreDataBlobRequest(&livekit.StoreDataBlobRequest{ + Blob: &livekit.DataBlob{ + Key: genericKey("blob-1"), + Contents: []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.dataBlob.GetAll()) + }) + + t.Run("stores a valid blob, notifies listener, and sends response", func(t *testing.T) { + p := newParticipantWithDataBlob(t, true, 0, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + listener := p.params.ParticipantListener.(*typesfakes.FakeLocalParticipantListener) + + key := genericKey("blob-1") + contents := []byte("definition-bytes") + blob := &livekit.DataBlob{Key: key, Contents: contents} + + p.HandleStoreDataBlobRequest(&livekit.StoreDataBlobRequest{ + RequestId: 42, + Blob: blob, + }) + + require.Equal(t, 1, sink.WriteMessageCallCount()) + msg := sink.WriteMessageArgsForCall(0).(*livekit.SignalResponse) + response, ok := msg.Message.(*livekit.SignalResponse_StoreDataBlobResponse) + require.True(t, ok, "expected SignalResponse_StoreDataBlobResponse, got %T", msg.Message) + require.Equal(t, uint32(42), response.StoreDataBlobResponse.RequestId) + require.Equal(t, key, response.StoreDataBlobResponse.Key) + + stored := p.dataBlob.Get(key) + require.NotNil(t, stored) + require.Equal(t, contents, stored.Contents) + + require.Equal(t, 1, listener.OnStoreDataBlobCallCount()) + gotParticipant, gotBlob := listener.OnStoreDataBlobArgsForCall(0) + require.Equal(t, p, gotParticipant) + require.Equal(t, blob, gotBlob) + }) +} + +func TestHandleGetDataBlobRequest(t *testing.T) { + t.Run("returns INVALID_REQUEST when key is missing", func(t *testing.T) { + p := newParticipantWithDataBlob(t, true, 0, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + p.HandleGetDataBlobRequest(&livekit.GetDataBlobRequest{ + 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 key is provided", func(t *testing.T) { + p := newParticipantWithDataBlob(t, true, 0, 0) + listener := p.params.ParticipantListener.(*typesfakes.FakeLocalParticipantListener) + + req := &livekit.GetDataBlobRequest{ + ParticipantIdentity: "other", + Key: genericKey("blob-1"), + } + p.HandleGetDataBlobRequest(req) + + require.Equal(t, 1, listener.OnGetDataBlobCallCount()) + gotParticipant, gotReq := listener.OnGetDataBlobArgsForCall(0) + require.Equal(t, p, gotParticipant) + require.Equal(t, req, gotReq) + }) +} + +func TestGetDataBlob(t *testing.T) { + p := newParticipantWithDataBlob(t, true, 0, 0) + + key := genericKey("blob-1") + require.Nil(t, p.GetDataBlob(key)) + + blob := &livekit.DataBlob{ + Key: key, + Contents: []byte("definition"), + } + p.dataBlob.Add(blob) + got := p.GetDataBlob(key) + require.NotNil(t, got) + require.Equal(t, key.String(), got.Key.String()) + require.Equal(t, []byte("definition"), got.Contents) +} + +func TestProcessGetDataBlobRequest(t *testing.T) { + t.Run("returns NOT_FOUND when publisher is nil", func(t *testing.T) { + p := newParticipantWithDataBlob(t, true, 0, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + p.ProcessGetDataBlobRequest(&livekit.GetDataBlobRequest{ + Key: genericKey("blob-1"), + }, 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 blob", func(t *testing.T) { + p := newParticipantWithDataBlob(t, true, 0, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + publisher := &typesfakes.FakeParticipant{} + publisher.GetDataBlobReturns(nil) + + req := &livekit.GetDataBlobRequest{ + Key: genericKey("blob-1"), + } + p.ProcessGetDataBlobRequest(req, publisher) + + require.Equal(t, 1, publisher.GetDataBlobCallCount()) + require.Equal(t, req.Key, publisher.GetDataBlobArgsForCall(0)) + + require.Equal(t, 1, sink.WriteMessageCallCount()) + rr := lastRequestResponse(t, sink, 0) + require.Equal(t, livekit.RequestResponse_NOT_FOUND, rr.Reason) + }) + + t.Run("sends blob response when publisher has a matching blob", func(t *testing.T) { + p := newParticipantWithDataBlob(t, true, 0, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + key := genericKey("blob-1") + blob := &livekit.DataBlob{ + Key: key, + Contents: []byte("definition-bytes"), + } + + publisher := &typesfakes.FakeParticipant{} + publisher.GetDataBlobReturns(blob) + + p.ProcessGetDataBlobRequest(&livekit.GetDataBlobRequest{ + RequestId: 42, + Key: key, + }, publisher) + + require.Equal(t, 1, sink.WriteMessageCallCount()) + msg := sink.WriteMessageArgsForCall(0).(*livekit.SignalResponse) + response, ok := msg.Message.(*livekit.SignalResponse_GetDataBlobResponse) + require.True(t, ok, "expected SignalResponse_GetDataBlobResponse, got %T", msg.Message) + require.Equal(t, uint32(42), response.GetDataBlobResponse.RequestId) + require.Equal(t, blob, response.GetDataBlobResponse.Blob) + }) +} diff --git a/pkg/rtc/participant_data_blob_test.go b/pkg/rtc/participant_data_blob_test.go new file mode 100644 index 000000000..a6d28fb20 --- /dev/null +++ b/pkg/rtc/participant_data_blob_test.go @@ -0,0 +1,175 @@ +// 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" + "sync" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" +) + +func newTestDataBlob() *ParticipantDataBlob { + return NewParticipantDataBlob(ParticipantDataBlobParams{ + Logger: logger.GetLogger(), + }) +} + +func genericKey(name string) *livekit.DataBlobKey { + return &livekit.DataBlobKey{ + Key: &livekit.DataBlobKey_Generic{ + Generic: name, + }, + } +} + +func TestParticipantDataBlob_AddAndGet(t *testing.T) { + a := newTestDataBlob() + + key := genericKey("blob-1") + contents := []byte("definition-bytes") + + a.Add(&livekit.DataBlob{Key: key, Contents: contents}) + + got := a.Get(key) + require.NotNil(t, got) + require.Equal(t, key.String(), got.Key.String()) + require.Equal(t, contents, got.Contents) +} + +func TestParticipantDataBlob_AddOverwrites(t *testing.T) { + a := newTestDataBlob() + + key := genericKey("blob-1") + a.Add(&livekit.DataBlob{Key: key, Contents: []byte("v1")}) + a.Add(&livekit.DataBlob{Key: key, Contents: []byte("v2")}) + + got := a.Get(key) + require.NotNil(t, got) + require.Equal(t, []byte("v2"), got.Contents) + + require.Len(t, a.GetAll(), 1) +} + +func TestParticipantDataBlob_DistinctKeys(t *testing.T) { + a := newTestDataBlob() + + key1 := genericKey("blob-1") + key2 := genericKey("blob-2") + + a.Add(&livekit.DataBlob{Key: key1, Contents: []byte("c1")}) + a.Add(&livekit.DataBlob{Key: key2, Contents: []byte("c2")}) + + got1 := a.Get(key1) + require.NotNil(t, got1) + require.Equal(t, []byte("c1"), got1.Contents) + + got2 := a.Get(key2) + require.NotNil(t, got2) + require.Equal(t, []byte("c2"), got2.Contents) + + require.Len(t, a.GetAll(), 2) +} + +func TestParticipantDataBlob_Delete(t *testing.T) { + a := newTestDataBlob() + + key := genericKey("blob-1") + a.Add(&livekit.DataBlob{Key: key, Contents: []byte("definition")}) + + a.Delete(key) + require.Nil(t, a.Get(key)) + require.Empty(t, a.GetAll()) + + // deleting a non-existent key is a no-op + a.Delete(key) + require.Empty(t, a.GetAll()) +} + +func TestParticipantDataBlob_NilKey(t *testing.T) { + a := newTestDataBlob() + + // nil key should be silently ignored, not panic + a.Add(&livekit.DataBlob{Contents: []byte("definition")}) + require.Empty(t, a.GetAll()) + + require.Nil(t, a.Get(nil)) + + a.Delete(nil) + require.Empty(t, a.GetAll()) +} + +func TestParticipantDataBlob_GetMissing(t *testing.T) { + a := newTestDataBlob() + + require.Nil(t, a.Get(genericKey("missing"))) +} + +func TestParticipantDataBlob_GetAllContents(t *testing.T) { + a := newTestDataBlob() + + key1 := genericKey("blob-1") + key2 := genericKey("blob-2") + + a.Add(&livekit.DataBlob{Key: key1, Contents: []byte("def-1")}) + a.Add(&livekit.DataBlob{Key: key2, Contents: []byte("def-2")}) + + all := a.GetAll() + require.Len(t, all, 2) + for _, db := range all { + switch key := db.Key.Key.(type) { + case *livekit.DataBlobKey_Generic: + switch key.Generic { + case "blob-1": + require.Equal(t, []byte("def-1"), db.Contents) + case "blob-2": + require.Equal(t, []byte("def-2"), db.Contents) + default: + require.Fail(t, "unexpected key", key.Generic) + } + default: + require.Fail(t, "unexpected key type", "Generic") + } + } +} + +func TestParticipantDataBlob_ConcurrentAccess(t *testing.T) { + a := newTestDataBlob() + + 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++ { + key := genericKey(fmt.Sprintf("blob-%d", g%8)) + a.Add(&livekit.DataBlob{Key: key, Contents: []byte("v")}) + _ = a.Get(key) + _ = a.GetAll() + if i%3 == 0 { + a.Delete(key) + } + } + }(g) + } + wg.Wait() +} diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go index 3fa388adc..8f5c3dd2b 100644 --- a/pkg/rtc/participant_signal.go +++ b/pkg/rtc/participant_signal.go @@ -368,3 +368,17 @@ func (p *ParticipantImpl) SendDataTrackSubscriberHandles(handles map[uint32]*liv SubHandles: handles, })) } + +func (p *ParticipantImpl) sendStoreDataBlobResponse(requestId uint32, key *livekit.DataBlobKey) error { + return p.signaller.WriteMessage(p.signalling.SignalStoreDataBlobResponse(&livekit.StoreDataBlobResponse{ + RequestId: requestId, + Key: key, + })) +} + +func (p *ParticipantImpl) sendGetDataBlobResponse(requestId uint32, dataBlob *livekit.DataBlob) error { + return p.signaller.WriteMessage(p.signalling.SignalGetDataBlobResponse(&livekit.GetDataBlobResponse{ + RequestId: requestId, + Blob: dataBlob, + })) +} diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 86ddcd05e..66e922ec3 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -1384,6 +1384,11 @@ func (r *Room) onUpdateDataSubscriptions(participant types.LocalParticipant, req } } +func (r *Room) onGetDataBlob(participant types.LocalParticipant, req *livekit.GetDataBlobRequest) { + publisher := r.GetParticipant(livekit.ParticipantIdentity(req.ParticipantIdentity)) + participant.ProcessGetDataBlobRequest(req, publisher) +} + func (r *Room) onLeave(p types.LocalParticipant, reason types.ParticipantCloseReason) { r.RemoveParticipant(p.Identity(), p.ID(), reason) } @@ -1996,6 +2001,13 @@ func (l *localParticipantListener) OnUpdateDataSubscriptions(p types.LocalPartic l.room.onUpdateDataSubscriptions(p, req) } +func (l *localParticipantListener) OnStoreDataBlob(_p types.LocalParticipant, _dataBlob *livekit.DataBlob) { +} + +func (l *localParticipantListener) OnGetDataBlob(p types.LocalParticipant, req *livekit.GetDataBlobRequest) { + l.room.onGetDataBlob(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..99c3a0acc 100644 --- a/pkg/rtc/signalling/interfaces.go +++ b/pkg/rtc/signalling/interfaces.go @@ -62,4 +62,6 @@ type ParticipantSignalling interface { SignalPublishDataTrackResponse(publishDataTrackResponse *livekit.PublishDataTrackResponse) proto.Message SignalUnpublishDataTrackResponse(unpublishDataTrackResponse *livekit.UnpublishDataTrackResponse) proto.Message SignalDataTrackSubscriberHandles(dataTrackSubscriberHandles *livekit.DataTrackSubscriberHandles) proto.Message + SignalStoreDataBlobResponse(storeDataBlobResponse *livekit.StoreDataBlobResponse) proto.Message + SignalGetDataBlobResponse(getDataBlobResponse *livekit.GetDataBlobResponse) proto.Message } diff --git a/pkg/rtc/signalling/signalhandler.go b/pkg/rtc/signalling/signalhandler.go index 7e69aeee2..740e65899 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_StoreDataBlobRequest: + s.params.Participant.HandleStoreDataBlobRequest(msg.StoreDataBlobRequest) + + case *livekit.SignalRequest_GetDataBlobRequest: + s.params.Participant.HandleGetDataBlobRequest(msg.GetDataBlobRequest) } return nil diff --git a/pkg/rtc/signalling/signalling.go b/pkg/rtc/signalling/signalling.go index fffc2ba29..c13cee1be 100644 --- a/pkg/rtc/signalling/signalling.go +++ b/pkg/rtc/signalling/signalling.go @@ -258,3 +258,19 @@ func (s *signalling) SignalDataTrackSubscriberHandles(dataTrackSubscriberHandles }, } } + +func (s *signalling) SignalStoreDataBlobResponse(storeDataBlobResponse *livekit.StoreDataBlobResponse) proto.Message { + return &livekit.SignalResponse{ + Message: &livekit.SignalResponse_StoreDataBlobResponse{ + StoreDataBlobResponse: storeDataBlobResponse, + }, + } +} + +func (s *signalling) SignalGetDataBlobResponse(getDataBlobResponse *livekit.GetDataBlobResponse) proto.Message { + return &livekit.SignalResponse{ + Message: &livekit.SignalResponse_GetDataBlobResponse{ + GetDataBlobResponse: getDataBlobResponse, + }, + } +} diff --git a/pkg/rtc/signalling/signallingunimplemented.go b/pkg/rtc/signalling/signallingunimplemented.go index dca48779f..8210c6aac 100644 --- a/pkg/rtc/signalling/signallingunimplemented.go +++ b/pkg/rtc/signalling/signallingunimplemented.go @@ -127,3 +127,11 @@ func (u *signallingUnimplemented) SignalUnpublishDataTrackResponse(unpublishData func (u *signallingUnimplemented) SignalDataTrackSubscriberHandles(dataTrackSubscriberHandles *livekit.DataTrackSubscriberHandles) proto.Message { return nil } + +func (u *signallingUnimplemented) SignalStoreDataBlobResponse(storeDataBlobResponse *livekit.StoreDataBlobResponse) proto.Message { + return nil +} + +func (u *signallingUnimplemented) SignalGetDataBlobResponse(getDataBlobResponse *livekit.GetDataBlobResponse) proto.Message { + return nil +} diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index ec10fb396..5d0c702ed 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 + + AddDataBlob(dataBlob *livekit.DataBlob) + GetDataBlob(key *livekit.DataBlobKey) *livekit.DataBlob } // ------------------------------------------------------- @@ -562,6 +565,9 @@ type LocalParticipant interface { HandlePublishDataTrackRequest(*livekit.PublishDataTrackRequest) HandleUnpublishDataTrackRequest(*livekit.UnpublishDataTrackRequest) HandleUpdateDataSubscription(*livekit.UpdateDataSubscription) + HandleStoreDataBlobRequest(*livekit.StoreDataBlobRequest) + HandleGetDataBlobRequest(*livekit.GetDataBlobRequest) + ProcessGetDataBlobRequest(*livekit.GetDataBlobRequest, Participant) HandleSignalMessage(msg proto.Message) error @@ -572,6 +578,8 @@ type LocalParticipant interface { ClearParticipantListener() GetNextSubscribedDataTrackHandle() uint16 + + GetAllDataBlob() []*livekit.DataBlob } // --------------------------------------------- @@ -621,6 +629,8 @@ type LocalParticipantListener interface { ) OnUpdateSubscriptionPermission(LocalParticipant, *livekit.SubscriptionPermission) error OnUpdateDataSubscriptions(LocalParticipant, *livekit.UpdateDataSubscription) + OnStoreDataBlob(LocalParticipant, *livekit.DataBlob) + OnGetDataBlob(LocalParticipant, *livekit.GetDataBlobRequest) OnSyncState(LocalParticipant, *livekit.SyncState) error OnSimulateScenario(LocalParticipant, *livekit.SimulateScenario) error OnLeave(LocalParticipant, ParticipantCloseReason) @@ -652,6 +662,10 @@ func (*NullLocalParticipantListener) OnUpdateSubscriptionPermission(LocalPartici } func (*NullLocalParticipantListener) OnUpdateDataSubscriptions(LocalParticipant, *livekit.UpdateDataSubscription) { } +func (*NullLocalParticipantListener) OnStoreDataBlob(LocalParticipant, *livekit.DataBlob) { +} +func (*NullLocalParticipantListener) OnGetDataBlob(LocalParticipant, *livekit.GetDataBlobRequest) { +} 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 9570a28c2..d4198c2de 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 } + AddDataBlobStub func(*livekit.DataBlob) + addDataBlobMutex sync.RWMutex + addDataBlobArgsForCall []struct { + arg1 *livekit.DataBlob + } AddOnCloseStub func(string, func(types.LocalParticipant)) addOnCloseMutex sync.RWMutex addOnCloseArgsForCall []struct { @@ -216,6 +221,16 @@ type FakeLocalParticipant struct { getAdaptiveStreamReturnsOnCall map[int]struct { result1 bool } + GetAllDataBlobStub func() []*livekit.DataBlob + getAllDataBlobMutex sync.RWMutex + getAllDataBlobArgsForCall []struct { + } + getAllDataBlobReturns struct { + result1 []*livekit.DataBlob + } + getAllDataBlobReturnsOnCall map[int]struct { + result1 []*livekit.DataBlob + } GetAnswerStub func() (webrtc.SessionDescription, uint32, error) getAnswerMutex sync.RWMutex getAnswerArgsForCall []struct { @@ -305,6 +320,17 @@ type FakeLocalParticipant struct { getCountryReturnsOnCall map[int]struct { result1 string } + GetDataBlobStub func(*livekit.DataBlobKey) *livekit.DataBlob + getDataBlobMutex sync.RWMutex + getDataBlobArgsForCall []struct { + arg1 *livekit.DataBlobKey + } + getDataBlobReturns struct { + result1 *livekit.DataBlob + } + getDataBlobReturnsOnCall map[int]struct { + result1 *livekit.DataBlob + } GetDataTrackTransportStub func() types.DataTrackTransport getDataTrackTransportMutex sync.RWMutex getDataTrackTransportArgsForCall []struct { @@ -576,6 +602,11 @@ type FakeLocalParticipant struct { handleAnswerArgsForCall []struct { arg1 *livekit.SessionDescription } + HandleGetDataBlobRequestStub func(*livekit.GetDataBlobRequest) + handleGetDataBlobRequestMutex sync.RWMutex + handleGetDataBlobRequestArgsForCall []struct { + arg1 *livekit.GetDataBlobRequest + } HandleICERestartSDPFragmentStub func(string) (string, error) handleICERestartSDPFragmentMutex sync.RWMutex handleICERestartSDPFragmentArgsForCall []struct { @@ -689,6 +720,11 @@ type FakeLocalParticipant struct { handleSimulateScenarioReturnsOnCall map[int]struct { result1 error } + HandleStoreDataBlobRequestStub func(*livekit.StoreDataBlobRequest) + handleStoreDataBlobRequestMutex sync.RWMutex + handleStoreDataBlobRequestArgsForCall []struct { + arg1 *livekit.StoreDataBlobRequest + } HandleSyncStateStub func(*livekit.SyncState) error handleSyncStateMutex sync.RWMutex handleSyncStateArgsForCall []struct { @@ -986,6 +1022,12 @@ type FakeLocalParticipant struct { arg2 chan string arg3 chan error } + ProcessGetDataBlobRequestStub func(*livekit.GetDataBlobRequest, types.Participant) + processGetDataBlobRequestMutex sync.RWMutex + processGetDataBlobRequestArgsForCall []struct { + arg1 *livekit.GetDataBlobRequest + arg2 types.Participant + } ProtocolVersionStub func() types.ProtocolVersion protocolVersionMutex sync.RWMutex protocolVersionArgsForCall []struct { @@ -1594,6 +1636,38 @@ func (fake *FakeLocalParticipant) ActiveAtReturnsOnCall(i int, result1 time.Time }{result1} } +func (fake *FakeLocalParticipant) AddDataBlob(arg1 *livekit.DataBlob) { + fake.addDataBlobMutex.Lock() + fake.addDataBlobArgsForCall = append(fake.addDataBlobArgsForCall, struct { + arg1 *livekit.DataBlob + }{arg1}) + stub := fake.AddDataBlobStub + fake.recordInvocation("AddDataBlob", []interface{}{arg1}) + fake.addDataBlobMutex.Unlock() + if stub != nil { + fake.AddDataBlobStub(arg1) + } +} + +func (fake *FakeLocalParticipant) AddDataBlobCallCount() int { + fake.addDataBlobMutex.RLock() + defer fake.addDataBlobMutex.RUnlock() + return len(fake.addDataBlobArgsForCall) +} + +func (fake *FakeLocalParticipant) AddDataBlobCalls(stub func(*livekit.DataBlob)) { + fake.addDataBlobMutex.Lock() + defer fake.addDataBlobMutex.Unlock() + fake.AddDataBlobStub = stub +} + +func (fake *FakeLocalParticipant) AddDataBlobArgsForCall(i int) *livekit.DataBlob { + fake.addDataBlobMutex.RLock() + defer fake.addDataBlobMutex.RUnlock() + argsForCall := fake.addDataBlobArgsForCall[i] + return argsForCall.arg1 +} + func (fake *FakeLocalParticipant) AddOnClose(arg1 string, arg2 func(types.LocalParticipant)) { fake.addOnCloseMutex.Lock() fake.addOnCloseArgsForCall = append(fake.addOnCloseArgsForCall, struct { @@ -2539,6 +2613,59 @@ func (fake *FakeLocalParticipant) GetAdaptiveStreamReturnsOnCall(i int, result1 }{result1} } +func (fake *FakeLocalParticipant) GetAllDataBlob() []*livekit.DataBlob { + fake.getAllDataBlobMutex.Lock() + ret, specificReturn := fake.getAllDataBlobReturnsOnCall[len(fake.getAllDataBlobArgsForCall)] + fake.getAllDataBlobArgsForCall = append(fake.getAllDataBlobArgsForCall, struct { + }{}) + stub := fake.GetAllDataBlobStub + fakeReturns := fake.getAllDataBlobReturns + fake.recordInvocation("GetAllDataBlob", []interface{}{}) + fake.getAllDataBlobMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) GetAllDataBlobCallCount() int { + fake.getAllDataBlobMutex.RLock() + defer fake.getAllDataBlobMutex.RUnlock() + return len(fake.getAllDataBlobArgsForCall) +} + +func (fake *FakeLocalParticipant) GetAllDataBlobCalls(stub func() []*livekit.DataBlob) { + fake.getAllDataBlobMutex.Lock() + defer fake.getAllDataBlobMutex.Unlock() + fake.GetAllDataBlobStub = stub +} + +func (fake *FakeLocalParticipant) GetAllDataBlobReturns(result1 []*livekit.DataBlob) { + fake.getAllDataBlobMutex.Lock() + defer fake.getAllDataBlobMutex.Unlock() + fake.GetAllDataBlobStub = nil + fake.getAllDataBlobReturns = struct { + result1 []*livekit.DataBlob + }{result1} +} + +func (fake *FakeLocalParticipant) GetAllDataBlobReturnsOnCall(i int, result1 []*livekit.DataBlob) { + fake.getAllDataBlobMutex.Lock() + defer fake.getAllDataBlobMutex.Unlock() + fake.GetAllDataBlobStub = nil + if fake.getAllDataBlobReturnsOnCall == nil { + fake.getAllDataBlobReturnsOnCall = make(map[int]struct { + result1 []*livekit.DataBlob + }) + } + fake.getAllDataBlobReturnsOnCall[i] = struct { + result1 []*livekit.DataBlob + }{result1} +} + func (fake *FakeLocalParticipant) GetAnswer() (webrtc.SessionDescription, uint32, error) { fake.getAnswerMutex.Lock() ret, specificReturn := fake.getAnswerReturnsOnCall[len(fake.getAnswerArgsForCall)] @@ -2983,6 +3110,67 @@ func (fake *FakeLocalParticipant) GetCountryReturnsOnCall(i int, result1 string) }{result1} } +func (fake *FakeLocalParticipant) GetDataBlob(arg1 *livekit.DataBlobKey) *livekit.DataBlob { + fake.getDataBlobMutex.Lock() + ret, specificReturn := fake.getDataBlobReturnsOnCall[len(fake.getDataBlobArgsForCall)] + fake.getDataBlobArgsForCall = append(fake.getDataBlobArgsForCall, struct { + arg1 *livekit.DataBlobKey + }{arg1}) + stub := fake.GetDataBlobStub + fakeReturns := fake.getDataBlobReturns + fake.recordInvocation("GetDataBlob", []interface{}{arg1}) + fake.getDataBlobMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) GetDataBlobCallCount() int { + fake.getDataBlobMutex.RLock() + defer fake.getDataBlobMutex.RUnlock() + return len(fake.getDataBlobArgsForCall) +} + +func (fake *FakeLocalParticipant) GetDataBlobCalls(stub func(*livekit.DataBlobKey) *livekit.DataBlob) { + fake.getDataBlobMutex.Lock() + defer fake.getDataBlobMutex.Unlock() + fake.GetDataBlobStub = stub +} + +func (fake *FakeLocalParticipant) GetDataBlobArgsForCall(i int) *livekit.DataBlobKey { + fake.getDataBlobMutex.RLock() + defer fake.getDataBlobMutex.RUnlock() + argsForCall := fake.getDataBlobArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeLocalParticipant) GetDataBlobReturns(result1 *livekit.DataBlob) { + fake.getDataBlobMutex.Lock() + defer fake.getDataBlobMutex.Unlock() + fake.GetDataBlobStub = nil + fake.getDataBlobReturns = struct { + result1 *livekit.DataBlob + }{result1} +} + +func (fake *FakeLocalParticipant) GetDataBlobReturnsOnCall(i int, result1 *livekit.DataBlob) { + fake.getDataBlobMutex.Lock() + defer fake.getDataBlobMutex.Unlock() + fake.GetDataBlobStub = nil + if fake.getDataBlobReturnsOnCall == nil { + fake.getDataBlobReturnsOnCall = make(map[int]struct { + result1 *livekit.DataBlob + }) + } + fake.getDataBlobReturnsOnCall[i] = struct { + result1 *livekit.DataBlob + }{result1} +} + func (fake *FakeLocalParticipant) GetDataTrackTransport() types.DataTrackTransport { fake.getDataTrackTransportMutex.Lock() ret, specificReturn := fake.getDataTrackTransportReturnsOnCall[len(fake.getDataTrackTransportArgsForCall)] @@ -4428,6 +4616,38 @@ func (fake *FakeLocalParticipant) HandleAnswerArgsForCall(i int) *livekit.Sessio return argsForCall.arg1 } +func (fake *FakeLocalParticipant) HandleGetDataBlobRequest(arg1 *livekit.GetDataBlobRequest) { + fake.handleGetDataBlobRequestMutex.Lock() + fake.handleGetDataBlobRequestArgsForCall = append(fake.handleGetDataBlobRequestArgsForCall, struct { + arg1 *livekit.GetDataBlobRequest + }{arg1}) + stub := fake.HandleGetDataBlobRequestStub + fake.recordInvocation("HandleGetDataBlobRequest", []interface{}{arg1}) + fake.handleGetDataBlobRequestMutex.Unlock() + if stub != nil { + fake.HandleGetDataBlobRequestStub(arg1) + } +} + +func (fake *FakeLocalParticipant) HandleGetDataBlobRequestCallCount() int { + fake.handleGetDataBlobRequestMutex.RLock() + defer fake.handleGetDataBlobRequestMutex.RUnlock() + return len(fake.handleGetDataBlobRequestArgsForCall) +} + +func (fake *FakeLocalParticipant) HandleGetDataBlobRequestCalls(stub func(*livekit.GetDataBlobRequest)) { + fake.handleGetDataBlobRequestMutex.Lock() + defer fake.handleGetDataBlobRequestMutex.Unlock() + fake.HandleGetDataBlobRequestStub = stub +} + +func (fake *FakeLocalParticipant) HandleGetDataBlobRequestArgsForCall(i int) *livekit.GetDataBlobRequest { + fake.handleGetDataBlobRequestMutex.RLock() + defer fake.handleGetDataBlobRequestMutex.RUnlock() + argsForCall := fake.handleGetDataBlobRequestArgsForCall[i] + return argsForCall.arg1 +} + func (fake *FakeLocalParticipant) HandleICERestartSDPFragment(arg1 string) (string, error) { fake.handleICERestartSDPFragmentMutex.Lock() ret, specificReturn := fake.handleICERestartSDPFragmentReturnsOnCall[len(fake.handleICERestartSDPFragmentArgsForCall)] @@ -5052,6 +5272,38 @@ func (fake *FakeLocalParticipant) HandleSimulateScenarioReturnsOnCall(i int, res }{result1} } +func (fake *FakeLocalParticipant) HandleStoreDataBlobRequest(arg1 *livekit.StoreDataBlobRequest) { + fake.handleStoreDataBlobRequestMutex.Lock() + fake.handleStoreDataBlobRequestArgsForCall = append(fake.handleStoreDataBlobRequestArgsForCall, struct { + arg1 *livekit.StoreDataBlobRequest + }{arg1}) + stub := fake.HandleStoreDataBlobRequestStub + fake.recordInvocation("HandleStoreDataBlobRequest", []interface{}{arg1}) + fake.handleStoreDataBlobRequestMutex.Unlock() + if stub != nil { + fake.HandleStoreDataBlobRequestStub(arg1) + } +} + +func (fake *FakeLocalParticipant) HandleStoreDataBlobRequestCallCount() int { + fake.handleStoreDataBlobRequestMutex.RLock() + defer fake.handleStoreDataBlobRequestMutex.RUnlock() + return len(fake.handleStoreDataBlobRequestArgsForCall) +} + +func (fake *FakeLocalParticipant) HandleStoreDataBlobRequestCalls(stub func(*livekit.StoreDataBlobRequest)) { + fake.handleStoreDataBlobRequestMutex.Lock() + defer fake.handleStoreDataBlobRequestMutex.Unlock() + fake.HandleStoreDataBlobRequestStub = stub +} + +func (fake *FakeLocalParticipant) HandleStoreDataBlobRequestArgsForCall(i int) *livekit.StoreDataBlobRequest { + fake.handleStoreDataBlobRequestMutex.RLock() + defer fake.handleStoreDataBlobRequestMutex.RUnlock() + argsForCall := fake.handleStoreDataBlobRequestArgsForCall[i] + return argsForCall.arg1 +} + func (fake *FakeLocalParticipant) HandleSyncState(arg1 *livekit.SyncState) error { fake.handleSyncStateMutex.Lock() ret, specificReturn := fake.handleSyncStateReturnsOnCall[len(fake.handleSyncStateArgsForCall)] @@ -6680,6 +6932,39 @@ func (fake *FakeLocalParticipant) PerformRpcArgsForCall(i int) (*livekit.Perform return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 } +func (fake *FakeLocalParticipant) ProcessGetDataBlobRequest(arg1 *livekit.GetDataBlobRequest, arg2 types.Participant) { + fake.processGetDataBlobRequestMutex.Lock() + fake.processGetDataBlobRequestArgsForCall = append(fake.processGetDataBlobRequestArgsForCall, struct { + arg1 *livekit.GetDataBlobRequest + arg2 types.Participant + }{arg1, arg2}) + stub := fake.ProcessGetDataBlobRequestStub + fake.recordInvocation("ProcessGetDataBlobRequest", []interface{}{arg1, arg2}) + fake.processGetDataBlobRequestMutex.Unlock() + if stub != nil { + fake.ProcessGetDataBlobRequestStub(arg1, arg2) + } +} + +func (fake *FakeLocalParticipant) ProcessGetDataBlobRequestCallCount() int { + fake.processGetDataBlobRequestMutex.RLock() + defer fake.processGetDataBlobRequestMutex.RUnlock() + return len(fake.processGetDataBlobRequestArgsForCall) +} + +func (fake *FakeLocalParticipant) ProcessGetDataBlobRequestCalls(stub func(*livekit.GetDataBlobRequest, types.Participant)) { + fake.processGetDataBlobRequestMutex.Lock() + defer fake.processGetDataBlobRequestMutex.Unlock() + fake.ProcessGetDataBlobRequestStub = stub +} + +func (fake *FakeLocalParticipant) ProcessGetDataBlobRequestArgsForCall(i int) (*livekit.GetDataBlobRequest, types.Participant) { + fake.processGetDataBlobRequestMutex.RLock() + defer fake.processGetDataBlobRequestMutex.RUnlock() + argsForCall := fake.processGetDataBlobRequestArgsForCall[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..4c0a3a07a 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant_listener.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant_listener.go @@ -42,6 +42,12 @@ type FakeLocalParticipantListener struct { arg1 types.Participant arg2 types.DataTrack } + OnGetDataBlobStub func(types.LocalParticipant, *livekit.GetDataBlobRequest) + onGetDataBlobMutex sync.RWMutex + onGetDataBlobArgsForCall []struct { + arg1 types.LocalParticipant + arg2 *livekit.GetDataBlobRequest + } OnLeaveStub func(types.LocalParticipant, types.ParticipantCloseReason) onLeaveMutex sync.RWMutex onLeaveArgsForCall []struct { @@ -82,6 +88,12 @@ type FakeLocalParticipantListener struct { onStateChangeArgsForCall []struct { arg1 types.LocalParticipant } + OnStoreDataBlobStub func(types.LocalParticipant, *livekit.DataBlob) + onStoreDataBlobMutex sync.RWMutex + onStoreDataBlobArgsForCall []struct { + arg1 types.LocalParticipant + arg2 *livekit.DataBlob + } OnSubscribeStatusChangedStub func(types.LocalParticipant, livekit.ParticipantID, bool) onSubscribeStatusChangedMutex sync.RWMutex onSubscribeStatusChangedArgsForCall []struct { @@ -331,6 +343,39 @@ func (fake *FakeLocalParticipantListener) OnDataTrackUnpublishedArgsForCall(i in return argsForCall.arg1, argsForCall.arg2 } +func (fake *FakeLocalParticipantListener) OnGetDataBlob(arg1 types.LocalParticipant, arg2 *livekit.GetDataBlobRequest) { + fake.onGetDataBlobMutex.Lock() + fake.onGetDataBlobArgsForCall = append(fake.onGetDataBlobArgsForCall, struct { + arg1 types.LocalParticipant + arg2 *livekit.GetDataBlobRequest + }{arg1, arg2}) + stub := fake.OnGetDataBlobStub + fake.recordInvocation("OnGetDataBlob", []interface{}{arg1, arg2}) + fake.onGetDataBlobMutex.Unlock() + if stub != nil { + fake.OnGetDataBlobStub(arg1, arg2) + } +} + +func (fake *FakeLocalParticipantListener) OnGetDataBlobCallCount() int { + fake.onGetDataBlobMutex.RLock() + defer fake.onGetDataBlobMutex.RUnlock() + return len(fake.onGetDataBlobArgsForCall) +} + +func (fake *FakeLocalParticipantListener) OnGetDataBlobCalls(stub func(types.LocalParticipant, *livekit.GetDataBlobRequest)) { + fake.onGetDataBlobMutex.Lock() + defer fake.onGetDataBlobMutex.Unlock() + fake.OnGetDataBlobStub = stub +} + +func (fake *FakeLocalParticipantListener) OnGetDataBlobArgsForCall(i int) (types.LocalParticipant, *livekit.GetDataBlobRequest) { + fake.onGetDataBlobMutex.RLock() + defer fake.onGetDataBlobMutex.RUnlock() + argsForCall := fake.onGetDataBlobArgsForCall[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 { @@ -556,6 +601,39 @@ func (fake *FakeLocalParticipantListener) OnStateChangeArgsForCall(i int) types. return argsForCall.arg1 } +func (fake *FakeLocalParticipantListener) OnStoreDataBlob(arg1 types.LocalParticipant, arg2 *livekit.DataBlob) { + fake.onStoreDataBlobMutex.Lock() + fake.onStoreDataBlobArgsForCall = append(fake.onStoreDataBlobArgsForCall, struct { + arg1 types.LocalParticipant + arg2 *livekit.DataBlob + }{arg1, arg2}) + stub := fake.OnStoreDataBlobStub + fake.recordInvocation("OnStoreDataBlob", []interface{}{arg1, arg2}) + fake.onStoreDataBlobMutex.Unlock() + if stub != nil { + fake.OnStoreDataBlobStub(arg1, arg2) + } +} + +func (fake *FakeLocalParticipantListener) OnStoreDataBlobCallCount() int { + fake.onStoreDataBlobMutex.RLock() + defer fake.onStoreDataBlobMutex.RUnlock() + return len(fake.onStoreDataBlobArgsForCall) +} + +func (fake *FakeLocalParticipantListener) OnStoreDataBlobCalls(stub func(types.LocalParticipant, *livekit.DataBlob)) { + fake.onStoreDataBlobMutex.Lock() + defer fake.onStoreDataBlobMutex.Unlock() + fake.OnStoreDataBlobStub = stub +} + +func (fake *FakeLocalParticipantListener) OnStoreDataBlobArgsForCall(i int) (types.LocalParticipant, *livekit.DataBlob) { + fake.onStoreDataBlobMutex.RLock() + defer fake.onStoreDataBlobMutex.RUnlock() + argsForCall := fake.onStoreDataBlobArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + func (fake *FakeLocalParticipantListener) OnSubscribeStatusChanged(arg1 types.LocalParticipant, arg2 livekit.ParticipantID, arg3 bool) { fake.onSubscribeStatusChangedMutex.Lock() fake.onSubscribeStatusChangedArgsForCall = append(fake.onSubscribeStatusChangedArgsForCall, struct { diff --git a/pkg/rtc/types/typesfakes/fake_participant.go b/pkg/rtc/types/typesfakes/fake_participant.go index 19498c9c3..652c8bf14 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 { + AddDataBlobStub func(*livekit.DataBlob) + addDataBlobMutex sync.RWMutex + addDataBlobArgsForCall []struct { + arg1 *livekit.DataBlob + } CanSkipBroadcastStub func() bool canSkipBroadcastMutex sync.RWMutex canSkipBroadcastArgsForCall []struct { @@ -78,6 +83,17 @@ type FakeParticipant struct { result1 float64 result2 bool } + GetDataBlobStub func(*livekit.DataBlobKey) *livekit.DataBlob + getDataBlobMutex sync.RWMutex + getDataBlobArgsForCall []struct { + arg1 *livekit.DataBlobKey + } + getDataBlobReturns struct { + result1 *livekit.DataBlob + } + getDataBlobReturnsOnCall map[int]struct { + result1 *livekit.DataBlob + } GetLoggerStub func() logger.Logger getLoggerMutex sync.RWMutex getLoggerArgsForCall []struct { @@ -361,6 +377,38 @@ type FakeParticipant struct { invocationsMutex sync.RWMutex } +func (fake *FakeParticipant) AddDataBlob(arg1 *livekit.DataBlob) { + fake.addDataBlobMutex.Lock() + fake.addDataBlobArgsForCall = append(fake.addDataBlobArgsForCall, struct { + arg1 *livekit.DataBlob + }{arg1}) + stub := fake.AddDataBlobStub + fake.recordInvocation("AddDataBlob", []interface{}{arg1}) + fake.addDataBlobMutex.Unlock() + if stub != nil { + fake.AddDataBlobStub(arg1) + } +} + +func (fake *FakeParticipant) AddDataBlobCallCount() int { + fake.addDataBlobMutex.RLock() + defer fake.addDataBlobMutex.RUnlock() + return len(fake.addDataBlobArgsForCall) +} + +func (fake *FakeParticipant) AddDataBlobCalls(stub func(*livekit.DataBlob)) { + fake.addDataBlobMutex.Lock() + defer fake.addDataBlobMutex.Unlock() + fake.AddDataBlobStub = stub +} + +func (fake *FakeParticipant) AddDataBlobArgsForCall(i int) *livekit.DataBlob { + fake.addDataBlobMutex.RLock() + defer fake.addDataBlobMutex.RUnlock() + argsForCall := fake.addDataBlobArgsForCall[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) GetDataBlob(arg1 *livekit.DataBlobKey) *livekit.DataBlob { + fake.getDataBlobMutex.Lock() + ret, specificReturn := fake.getDataBlobReturnsOnCall[len(fake.getDataBlobArgsForCall)] + fake.getDataBlobArgsForCall = append(fake.getDataBlobArgsForCall, struct { + arg1 *livekit.DataBlobKey + }{arg1}) + stub := fake.GetDataBlobStub + fakeReturns := fake.getDataBlobReturns + fake.recordInvocation("GetDataBlob", []interface{}{arg1}) + fake.getDataBlobMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeParticipant) GetDataBlobCallCount() int { + fake.getDataBlobMutex.RLock() + defer fake.getDataBlobMutex.RUnlock() + return len(fake.getDataBlobArgsForCall) +} + +func (fake *FakeParticipant) GetDataBlobCalls(stub func(*livekit.DataBlobKey) *livekit.DataBlob) { + fake.getDataBlobMutex.Lock() + defer fake.getDataBlobMutex.Unlock() + fake.GetDataBlobStub = stub +} + +func (fake *FakeParticipant) GetDataBlobArgsForCall(i int) *livekit.DataBlobKey { + fake.getDataBlobMutex.RLock() + defer fake.getDataBlobMutex.RUnlock() + argsForCall := fake.getDataBlobArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeParticipant) GetDataBlobReturns(result1 *livekit.DataBlob) { + fake.getDataBlobMutex.Lock() + defer fake.getDataBlobMutex.Unlock() + fake.GetDataBlobStub = nil + fake.getDataBlobReturns = struct { + result1 *livekit.DataBlob + }{result1} +} + +func (fake *FakeParticipant) GetDataBlobReturnsOnCall(i int, result1 *livekit.DataBlob) { + fake.getDataBlobMutex.Lock() + defer fake.getDataBlobMutex.Unlock() + fake.GetDataBlobStub = nil + if fake.getDataBlobReturnsOnCall == nil { + fake.getDataBlobReturnsOnCall = make(map[int]struct { + result1 *livekit.DataBlob + }) + } + fake.getDataBlobReturnsOnCall[i] = struct { + result1 *livekit.DataBlob + }{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 13ca67493..ce850df6e 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -538,6 +538,7 @@ func (r *RoomManager) StartSession( FireOnTrackBySdp: true, UseSinglePeerConnection: pi.UseSinglePeerConnection, EnableDataTracks: r.config.EnableDataTracks, + EnableParticipantDataBlob: r.config.EnableParticipantDataBlob, EnableRTPStreamRestartDetection: r.config.RTC.EnableRTPStreamRestartDetection, }) if err != nil { diff --git a/pkg/sfu/datachannel/datachannel_writer_test.go b/pkg/sfu/datachannel/datachannel_writer_test.go index 9b99b8929..3bdcefbf4 100644 --- a/pkg/sfu/datachannel/datachannel_writer_test.go +++ b/pkg/sfu/datachannel/datachannel_writer_test.go @@ -2,10 +2,11 @@ package datachannel import ( "context" - "sync/atomic" "testing" "time" + "go.uber.org/atomic" + "github.com/pion/datachannel" "github.com/pion/transport/v4/deadline" "github.com/stretchr/testify/require" 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..21f4294c6 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,129 @@ func TestCloseDisconnectedParticipantOnSignalClose(t *testing.T) { }) } } + +func TestMultiNodeDataBlob(t *testing.T) { + if testing.Short() { + t.SkipNow() + return + } + + _, _, finish := setupMultiNodeTestWithConfig("TestMultiNodeDataBlob", func(c *config.Config) { + c.EnableParticipantDataBlob = true + c.Limit.MaxDataBlobSize = 1024 + }) + defer finish() + + for _, testRTCServicePath := range testRTCServicePaths { + t.Run(fmt.Sprintf("testRTCServicePath=%s", testRTCServicePath.String()), func(t *testing.T) { + pubCapture := &dataBlobCapture{} + subCapture := &dataBlobCapture{} + + // 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 "" + }) + + key := &livekit.DataBlobKey{ + Key: &livekit.DataBlobKey_Generic{ + Generic: "blob-multinode", + }, + } + contents := []byte("multinode-content") + + require.NoError(t, pub.SendRequest(&livekit.SignalRequest{ + Message: &livekit.SignalRequest_StoreDataBlobRequest{ + StoreDataBlobRequest: &livekit.StoreDataBlobRequest{ + RequestId: 1, + Blob: &livekit.DataBlob{ + Key: key, + Contents: contents, + }, + }, + }, + })) + + testutils.WithTimeout(t, func() string { + resp := pubCapture.takeStoreResponse() + if resp == nil { + return "publisher did not receive store response" + } + if resp.RequestId != 1 { + return fmt.Sprintf("expected store response request id 1, got %d", resp.RequestId) + } + if resp.Key == nil { + return "store response missing key" + } + if resp.Key.String() != key.String() { + return fmt.Sprintf("expected stored blob key %s, got %s", key.String(), resp.Key.String()) + } + return "" + }) + require.Equal(t, 0, pubCapture.requestResponseCount(), "publisher should not receive an error response on success") + + // subscriber on a different node asks for the blob; the request routes + // across nodes to the publisher. + require.NoError(t, sub.SendRequest(&livekit.SignalRequest{ + Message: &livekit.SignalRequest_GetDataBlobRequest{ + GetDataBlobRequest: &livekit.GetDataBlobRequest{ + ParticipantIdentity: "pub", + Key: key, + }, + }, + })) + + testutils.WithTimeout(t, func() string { + resp := subCapture.takeBlobResponse() + if resp == nil { + return "subscriber did not receive blob response" + } + if resp.Blob == nil { + return "blob response missing blob" + } + if resp.Blob.Key.String() != key.String() { + return fmt.Sprintf("expected data blob key %s, got %s", key.String(), resp.Blob.Key.String()) + } + if string(resp.Blob.Contents) != string(contents) { + return fmt.Sprintf("expected contents %q, got %q", contents, resp.Blob.Contents) + } + return "" + }) + + // requesting an unknown publisher identity should return NOT_FOUND + require.NoError(t, sub.SendRequest(&livekit.SignalRequest{ + Message: &livekit.SignalRequest_GetDataBlobRequest{ + GetDataBlobRequest: &livekit.GetDataBlobRequest{ + ParticipantIdentity: "unknown-publisher", + Key: key, + }, + }, + })) + + 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 cf09473f3..6d88f4d8a 100644 --- a/test/singlenode_test.go +++ b/test/singlenode_test.go @@ -1526,3 +1526,299 @@ func TestTurnAuthFailure(t *testing.T) { }) } } + +// dataBlobCapture buffers RequestResponse, StoreDataBlobResponse, and GetDataBlobResponse messages +// sent to a test client so they can be asserted on. Other messages flow through to the +// default handler. +type dataBlobCapture struct { + mu sync.Mutex + requestResponses []*livekit.RequestResponse + storeResponses []*livekit.StoreDataBlobResponse + blobResponses []*livekit.GetDataBlobResponse +} + +func (c *dataBlobCapture) 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_StoreDataBlobResponse: + c.mu.Lock() + c.storeResponses = append(c.storeResponses, m.StoreDataBlobResponse) + c.mu.Unlock() + case *livekit.SignalResponse_GetDataBlobResponse: + c.mu.Lock() + c.blobResponses = append(c.blobResponses, m.GetDataBlobResponse) + c.mu.Unlock() + } + return next(msg) + } +} + +func (c *dataBlobCapture) 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 *dataBlobCapture) takeStoreResponse() *livekit.StoreDataBlobResponse { + c.mu.Lock() + defer c.mu.Unlock() + if len(c.storeResponses) == 0 { + return nil + } + sr := c.storeResponses[0] + c.storeResponses = c.storeResponses[1:] + return sr +} + +func (c *dataBlobCapture) takeBlobResponse() *livekit.GetDataBlobResponse { + c.mu.Lock() + defer c.mu.Unlock() + if len(c.blobResponses) == 0 { + return nil + } + sr := c.blobResponses[0] + c.blobResponses = c.blobResponses[1:] + return sr +} + +func (c *dataBlobCapture) requestResponseCount() int { + c.mu.Lock() + defer c.mu.Unlock() + return len(c.requestResponses) +} + +func setupDataBlobServer(t *testing.T, name string, enable bool) (*service.LivekitServer, func()) { + logger.Infow("----------------STARTING TEST----------------", "test", name) + s := createSingleNodeServer(func(c *config.Config) { + c.EnableParticipantDataBlob = enable + c.Limit.MaxDataBlobSize = 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 TestSingleNodeDataBlob(t *testing.T) { + if testing.Short() { + t.SkipNow() + return + } + + _, finish := setupDataBlobServer(t, "TestSingleNodeDataBlob", true) + defer finish() + + for _, testRTCServicePath := range testRTCServicePaths { + t.Run(fmt.Sprintf("testRTCServicePath=%s", testRTCServicePath.String()), func(t *testing.T) { + pubCapture := &dataBlobCapture{} + subCapture := &dataBlobCapture{} + + 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) + + key := &livekit.DataBlobKey{ + Key: &livekit.DataBlobKey_Generic{ + Generic: "blob-1", + }, + } + contents := []byte("definition-bytes") + + // publisher stores a blob + require.NoError(t, pub.SendRequest(&livekit.SignalRequest{ + Message: &livekit.SignalRequest_StoreDataBlobRequest{ + StoreDataBlobRequest: &livekit.StoreDataBlobRequest{ + RequestId: 1, + Blob: &livekit.DataBlob{ + Key: key, + Contents: contents, + }, + }, + }, + })) + + testutils.WithTimeout(t, func() string { + resp := pubCapture.takeStoreResponse() + if resp == nil { + return "publisher did not receive store response" + } + if resp.RequestId != 1 { + return fmt.Sprintf("expected store response request id 1, got %d", resp.RequestId) + } + if resp.Key == nil { + return "store response missing key" + } + if resp.Key.String() != key.String() { + return fmt.Sprintf("expected stored blob key %s, got %s", key.String(), resp.Key.String()) + } + return "" + }) + require.Equal(t, 0, pubCapture.requestResponseCount(), "publisher should not receive an error response on success") + + // subscriber asks for the blob + require.NoError(t, sub.SendRequest(&livekit.SignalRequest{ + Message: &livekit.SignalRequest_GetDataBlobRequest{ + GetDataBlobRequest: &livekit.GetDataBlobRequest{ + ParticipantIdentity: "pub", + Key: key, + }, + }, + })) + + testutils.WithTimeout(t, func() string { + resp := subCapture.takeBlobResponse() + if resp == nil { + return "subscriber did not receive blob response" + } + if resp.Blob == nil { + return "blob response missing blob" + } + if resp.Blob.Key.String() != key.String() { + return fmt.Sprintf("expected blob key %s, got %s", key.String(), resp.Blob.Key.String()) + } + if string(resp.Blob.Contents) != string(contents) { + return fmt.Sprintf("expected contents %q, got %q", contents, resp.Blob.Contents) + } + return "" + }) + + // subscriber asks for an unknown blob on a known publisher + require.NoError(t, sub.SendRequest(&livekit.SignalRequest{ + Message: &livekit.SignalRequest_GetDataBlobRequest{ + GetDataBlobRequest: &livekit.GetDataBlobRequest{ + ParticipantIdentity: "pub", + Key: &livekit.DataBlobKey{ + Key: &livekit.DataBlobKey_Generic{ + Generic: "does-not-exist", + }, + }, + }, + }, + })) + + testutils.WithTimeout(t, func() string { + rr := subCapture.takeRequestResponse() + if rr == nil { + return "subscriber did not receive RequestResponse for missing blob" + } + if rr.Reason != livekit.RequestResponse_NOT_FOUND { + return fmt.Sprintf("expected NOT_FOUND, got %s", rr.Reason) + } + return "" + }) + + // subscriber asks for a blob on an unknown publisher identity + require.NoError(t, sub.SendRequest(&livekit.SignalRequest{ + Message: &livekit.SignalRequest_GetDataBlobRequest{ + GetDataBlobRequest: &livekit.GetDataBlobRequest{ + ParticipantIdentity: "unknown-publisher", + Key: key, + }, + }, + })) + + 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 blob (empty key) + require.NoError(t, pub.SendRequest(&livekit.SignalRequest{ + Message: &livekit.SignalRequest_StoreDataBlobRequest{ + StoreDataBlobRequest: &livekit.StoreDataBlobRequest{ + Blob: &livekit.DataBlob{ + Contents: contents, + }, + }, + }, + })) + + 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 TestSingleNodeDataBlobDisabled(t *testing.T) { + if testing.Short() { + t.SkipNow() + return + } + + _, finish := setupDataBlobServer(t, "TestSingleNodeDataBlobDisabled", false) + defer finish() + + for _, testRTCServicePath := range testRTCServicePaths { + t.Run(fmt.Sprintf("testRTCServicePath=%s", testRTCServicePath.String()), func(t *testing.T) { + pubCapture := &dataBlobCapture{} + 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_StoreDataBlobRequest{ + StoreDataBlobRequest: &livekit.StoreDataBlobRequest{ + Blob: &livekit.DataBlob{ + Key: &livekit.DataBlobKey{ + Key: &livekit.DataBlobKey_Generic{ + Generic: "blob-1", + }, + }, + Contents: []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 "" + }) + }) + } +}