mirror of
https://github.com/livekit/livekit.git
synced 2026-08-29 07:39:09 +00:00
Add support for data blob (a. k. a. async participant attributes) (#4619)
* 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. * clean up * get full definitions, not just ids * listener OnDataTrackSchema * name length config * data blob * deps * static check * Add missing request ID * Update protocol commit * Wire up StoreDataBlobResponse * Pass request ID through in GetDataBlobResponse * deps * atomic * sctp at 1.9.5 * remove proto clone --------- Co-authored-by: Jacob Gelman <3182119+ladvoc@users.noreply.github.com>
This commit is contained in:
Raja Subramanian
co-authored by
Jacob Gelman
Jacob Gelman
parent
0cf53e2f0d
commit
1faab0c48e
@@ -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
|
||||
)
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
// -------------------------------
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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)
|
||||
})
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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,
|
||||
}))
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)]
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)]
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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 ""
|
||||
})
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 ""
|
||||
})
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user