dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-29 09:49:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

Delete v2 signalling (#3835)

Browse Source 
* Delete v2 signalling

* deps

* mage generate

* deelte unused file
This commit is contained in:
Raja Subramanian
2025-08-06 09:08:09 +05:30
committed by GitHub
parent 1fe337160a
commit 34a491309f
34 changed files with 156 additions and 2664 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+4 -4
View File
@@ -23,7 +23,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-20250519131108-fb90f5acfded
github.com/livekit/protocol v1.39.4-0.20250802093152-1b7e38ef9c96
github.com/livekit/protocol v1.39.4-0.20250806031641-1edabe8e86df
github.com/livekit/psrpc v0.6.1-0.20250726180611-3915e005e741
github.com/mackerelio/go-osstat v0.2.5
github.com/magefile/mage v1.15.0
@@ -44,7 +44,7 @@ require (
github.com/pion/webrtc/v4 v4.1.3
github.com/pkg/errors v0.9.1
github.com/prometheus/client_golang v1.22.0
github.com/redis/go-redis/v9 v9.11.0
github.com/redis/go-redis/v9 v9.12.0
github.com/rs/cors v1.11.1
github.com/stretchr/testify v1.10.0
github.com/thoas/go-funk v0.9.3
@@ -135,8 +135,8 @@ require (
golang.org/x/sys v0.34.0 // indirect
golang.org/x/text v0.27.0 // indirect
golang.org/x/tools v0.35.0 // indirect
google.golang.org/genproto/googleapis/api v0.0.0-20250728155136-f173205681a0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20250728155136-f173205681a0 // indirect
google.golang.org/genproto/googleapis/api v0.0.0-20250804133106-a7a43d27e69b // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20250804133106-a7a43d27e69b // indirect
google.golang.org/grpc v1.74.2 // indirect
gopkg.in/yaml.v2 v2.4.0 // indirect
)
go.sum
+8 -8
View File
@@ -167,8 +167,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-20250519131108-fb90f5acfded h1:ylZPdnlX1RW9Z15SD4mp87vT2D2shsk0hpLJwSPcq3g=
github.com/livekit/mediatransportutil v0.0.0-20250519131108-fb90f5acfded/go.mod h1:mSNtYzSf6iY9xM3UX42VEI+STHvMgHmrYzEHPcdhB8A=
github.com/livekit/protocol v1.39.4-0.20250802093152-1b7e38ef9c96 h1:PeGSL1wZKCU+Yizlbvmw60+NHnOWcSE19crqqnAeJgw=
github.com/livekit/protocol v1.39.4-0.20250802093152-1b7e38ef9c96/go.mod h1:YlgUxAegtU8jZ0tVXoIV/4fHeHqqLvS+6JnPKDbpFPU=
github.com/livekit/protocol v1.39.4-0.20250806031641-1edabe8e86df h1:3YB9qvVAPK0SNWDngETtE7UL75xAygZw07DtsYbKSNk=
github.com/livekit/protocol v1.39.4-0.20250806031641-1edabe8e86df/go.mod h1:YlgUxAegtU8jZ0tVXoIV/4fHeHqqLvS+6JnPKDbpFPU=
github.com/livekit/psrpc v0.6.1-0.20250726180611-3915e005e741 h1:KKL1u94l6dF9u4cBwnnfozk27GH1txWy2SlvkfgmzoY=
github.com/livekit/psrpc v0.6.1-0.20250726180611-3915e005e741/go.mod h1:AuDC5uOoEjQJEc69v4Li3t77Ocz0e0NdjQEuFfO+vfk=
github.com/mackerelio/go-osstat v0.2.5 h1:+MqTbZUhoIt4m8qzkVoXUJg1EuifwlAJSk4Yl2GXh+o=
@@ -276,8 +276,8 @@ github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzM
github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is=
github.com/puzpuzpuz/xsync/v3 v3.5.1 h1:GJYJZwO6IdxN/IKbneznS6yPkVC+c3zyY/j19c++5Fg=
github.com/puzpuzpuz/xsync/v3 v3.5.1/go.mod h1:VjzYrABPabuM4KyBh1Ftq6u8nhwY5tBPKP9jpmh0nnA=
github.com/redis/go-redis/v9 v9.11.0 h1:E3S08Gl/nJNn5vkxd2i78wZxWAPNZgUNTp8WIJUAiIs=
github.com/redis/go-redis/v9 v9.11.0/go.mod h1:huWgSWd8mW6+m0VPhJjSSQ+d6Nh1VICQ6Q5lHuCH/Iw=
github.com/redis/go-redis/v9 v9.12.0 h1:XlVPGlflh4nxfhsNXPA8Qp6EmEfTo0rp8oaBzPipXnU=
github.com/redis/go-redis/v9 v9.12.0/go.mod h1:huWgSWd8mW6+m0VPhJjSSQ+d6Nh1VICQ6Q5lHuCH/Iw=
github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M=
github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA=
@@ -471,10 +471,10 @@ golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8T
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
google.golang.org/genproto/googleapis/api v0.0.0-20250728155136-f173205681a0 h1:0UOBWO4dC+e51ui0NFKSPbkHHiQ4TmrEfEZMLDyRmY8=
google.golang.org/genproto/googleapis/api v0.0.0-20250728155136-f173205681a0/go.mod h1:8ytArBbtOy2xfht+y2fqKd5DRDJRUQhqbyEnQ4bDChs=
google.golang.org/genproto/googleapis/rpc v0.0.0-20250728155136-f173205681a0 h1:MAKi5q709QWfnkkpNQ0M12hYJ1+e8qYVDyowc4U1XZM=
google.golang.org/genproto/googleapis/rpc v0.0.0-20250728155136-f173205681a0/go.mod h1:qQ0YXyHHx3XkvlzUtpXDkS29lDSafHMZBAZDc03LQ3A=
google.golang.org/genproto/googleapis/api v0.0.0-20250804133106-a7a43d27e69b h1:ULiyYQ0FdsJhwwZUwbaXpZF5yUE3h+RA+gxvBu37ucc=
google.golang.org/genproto/googleapis/api v0.0.0-20250804133106-a7a43d27e69b/go.mod h1:oDOGiMSXHL4sDTJvFvIB9nRQCGdLP1o/iVaqQK8zB+M=
google.golang.org/genproto/googleapis/rpc v0.0.0-20250804133106-a7a43d27e69b h1:zPKJod4w6F1+nRGDI9ubnXYhU9NSWoFAijkHkUXeTK8=
google.golang.org/genproto/googleapis/rpc v0.0.0-20250804133106-a7a43d27e69b/go.mod h1:qQ0YXyHHx3XkvlzUtpXDkS29lDSafHMZBAZDc03LQ3A=
google.golang.org/grpc v1.74.2 h1:WoosgB65DlWVC9FqI82dGsZhWFNBSLjQ84bjROOpMu4=
google.golang.org/grpc v1.74.2/go.mod h1:CtQ+BGjaAIXHs/5YS3i473GqwBBa1zGQNevxdeBEXrM=
google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY=
pkg/routing/interfaces.go
-8
View File
@@ -160,14 +160,6 @@ type MessageRouter interface {
roomName livekit.RoomName,
pi ParticipantInit,
) (res StartParticipantSignalResults, err error)
// HandleParticipantConnectRequest handles connection request from participant
HandleParticipantConnectRequest(
ctx context.Context,
roomName livekit.RoomName,
participantIdentity livekit.ParticipantIdentity,
rscr *rpc.RelaySignalv2ConnectRequest,
) (resp *rpc.RelaySignalv2ConnectResponse, err error)
}
func CreateRouter(
pkg/routing/localrouter.go
-29
View File
@@ -24,7 +24,6 @@ import (
"github.com/livekit/livekit-server/pkg/config"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
"github.com/livekit/protocol/rpc"
)
var _ Router = (*LocalRouter)(nil)
@@ -128,34 +127,6 @@ func (r *LocalRouter) StartParticipantSignalWithNodeID(ctx context.Context, room
return
}
func (r *LocalRouter) HandleParticipantConnectRequest(
ctx context.Context,
roomName livekit.RoomName,
participantIdentity livekit.ParticipantIdentity,
rscr *rpc.RelaySignalv2ConnectRequest,
) (*rpc.RelaySignalv2ConnectResponse, error) {
return r.HandleParticipantConnectRequestWithNodeID(ctx, roomName, participantIdentity, rscr, r.currentNode.NodeID())
}
func (r *LocalRouter) HandleParticipantConnectRequestWithNodeID(
ctx context.Context,
roomName livekit.RoomName,
participantIdentity livekit.ParticipantIdentity,
rscr *rpc.RelaySignalv2ConnectRequest,
nodeID livekit.NodeID,
) (*rpc.RelaySignalv2ConnectResponse, error) {
resp, err := r.signalClient.HandleParticipantConnectRequest(ctx, roomName, participantIdentity, nodeID, rscr)
if err != nil {
logger.Errorw(
"could not handle new participant", err,
"room", roomName,
"participant", participantIdentity,
// SIGNALLING-V2-TODO "connID", connectionID,
)
}
return resp, err
}
func (r *LocalRouter) Start() error {
if r.isStarted.Swap(true) {
return nil
pkg/routing/redisrouter.go
-21
View File
@@ -169,27 +169,6 @@ func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName livek
return r.StartParticipantSignalWithNodeID(ctx, roomName, pi, livekit.NodeID(rtcNode.Id))
}
// HandleParticipantConnectRequest sends participant connect request to the RTC node for the room
func (r *RedisRouter) HandleParticipantConnectRequest(
ctx context.Context,
roomName livekit.RoomName,
participantIdentity livekit.ParticipantIdentity,
rscr *rpc.RelaySignalv2ConnectRequest,
) (*rpc.RelaySignalv2ConnectResponse, error) {
rtcNode, err := r.GetNodeForRoom(ctx, roomName)
if err != nil {
return nil, err
}
return r.HandleParticipantConnectRequestWithNodeID(
ctx,
roomName,
participantIdentity,
rscr,
livekit.NodeID(rtcNode.Id),
)
}
func (r *RedisRouter) Start() error {
if r.isStarted.Swap(true) {
return nil
pkg/routing/routingfakes/fake_router.go
-86
View File
@@ -7,7 +7,6 @@ import (
"github.com/livekit/livekit-server/pkg/routing"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/rpc"
)
type FakeRouter struct {
@@ -65,22 +64,6 @@ type FakeRouter struct {
getRegionReturnsOnCall map[int]struct {
result1 string
}
HandleParticipantConnectRequestStub func(context.Context, livekit.RoomName, livekit.ParticipantIdentity, *rpc.RelaySignalv2ConnectRequest) (*rpc.RelaySignalv2ConnectResponse, error)
handleParticipantConnectRequestMutex sync.RWMutex
handleParticipantConnectRequestArgsForCall []struct {
arg1 context.Context
arg2 livekit.RoomName
arg3 livekit.ParticipantIdentity
arg4 *rpc.RelaySignalv2ConnectRequest
}
handleParticipantConnectRequestReturns struct {
result1 *rpc.RelaySignalv2ConnectResponse
result2 error
}
handleParticipantConnectRequestReturnsOnCall map[int]struct {
result1 *rpc.RelaySignalv2ConnectResponse
result2 error
}
ListNodesStub func() ([]*livekit.Node, error)
listNodesMutex sync.RWMutex
listNodesArgsForCall []struct {
@@ -438,73 +421,6 @@ func (fake *FakeRouter) GetRegionReturnsOnCall(i int, result1 string) {
}{result1}
}
func (fake *FakeRouter) HandleParticipantConnectRequest(arg1 context.Context, arg2 livekit.RoomName, arg3 livekit.ParticipantIdentity, arg4 *rpc.RelaySignalv2ConnectRequest) (*rpc.RelaySignalv2ConnectResponse, error) {
fake.handleParticipantConnectRequestMutex.Lock()
ret, specificReturn := fake.handleParticipantConnectRequestReturnsOnCall[len(fake.handleParticipantConnectRequestArgsForCall)]
fake.handleParticipantConnectRequestArgsForCall = append(fake.handleParticipantConnectRequestArgsForCall, struct {
arg1 context.Context
arg2 livekit.RoomName
arg3 livekit.ParticipantIdentity
arg4 *rpc.RelaySignalv2ConnectRequest
}{arg1, arg2, arg3, arg4})
stub := fake.HandleParticipantConnectRequestStub
fakeReturns := fake.handleParticipantConnectRequestReturns
fake.recordInvocation("HandleParticipantConnectRequest", []interface{}{arg1, arg2, arg3, arg4})
fake.handleParticipantConnectRequestMutex.Unlock()
if stub != nil {
return stub(arg1, arg2, arg3, arg4)
}
if specificReturn {
return ret.result1, ret.result2
}
return fakeReturns.result1, fakeReturns.result2
}
func (fake *FakeRouter) HandleParticipantConnectRequestCallCount() int {
fake.handleParticipantConnectRequestMutex.RLock()
defer fake.handleParticipantConnectRequestMutex.RUnlock()
return len(fake.handleParticipantConnectRequestArgsForCall)
}
func (fake *FakeRouter) HandleParticipantConnectRequestCalls(stub func(context.Context, livekit.RoomName, livekit.ParticipantIdentity, *rpc.RelaySignalv2ConnectRequest) (*rpc.RelaySignalv2ConnectResponse, error)) {
fake.handleParticipantConnectRequestMutex.Lock()
defer fake.handleParticipantConnectRequestMutex.Unlock()
fake.HandleParticipantConnectRequestStub = stub
}
func (fake *FakeRouter) HandleParticipantConnectRequestArgsForCall(i int) (context.Context, livekit.RoomName, livekit.ParticipantIdentity, *rpc.RelaySignalv2ConnectRequest) {
fake.handleParticipantConnectRequestMutex.RLock()
defer fake.handleParticipantConnectRequestMutex.RUnlock()
argsForCall := fake.handleParticipantConnectRequestArgsForCall[i]
return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4
}
func (fake *FakeRouter) HandleParticipantConnectRequestReturns(result1 *rpc.RelaySignalv2ConnectResponse, result2 error) {
fake.handleParticipantConnectRequestMutex.Lock()
defer fake.handleParticipantConnectRequestMutex.Unlock()
fake.HandleParticipantConnectRequestStub = nil
fake.handleParticipantConnectRequestReturns = struct {
result1 *rpc.RelaySignalv2ConnectResponse
result2 error
}{result1, result2}
}
func (fake *FakeRouter) HandleParticipantConnectRequestReturnsOnCall(i int, result1 *rpc.RelaySignalv2ConnectResponse, result2 error) {
fake.handleParticipantConnectRequestMutex.Lock()
defer fake.handleParticipantConnectRequestMutex.Unlock()
fake.HandleParticipantConnectRequestStub = nil
if fake.handleParticipantConnectRequestReturnsOnCall == nil {
fake.handleParticipantConnectRequestReturnsOnCall = make(map[int]struct {
result1 *rpc.RelaySignalv2ConnectResponse
result2 error
})
}
fake.handleParticipantConnectRequestReturnsOnCall[i] = struct {
result1 *rpc.RelaySignalv2ConnectResponse
result2 error
}{result1, result2}
}
func (fake *FakeRouter) ListNodes() ([]*livekit.Node, error) {
fake.listNodesMutex.Lock()
ret, specificReturn := fake.listNodesReturnsOnCall[len(fake.listNodesArgsForCall)]
@@ -939,8 +855,6 @@ func (fake *FakeRouter) Invocations() map[string][][]interface{} {
defer fake.getNodeForRoomMutex.RUnlock()
fake.getRegionMutex.RLock()
defer fake.getRegionMutex.RUnlock()
fake.handleParticipantConnectRequestMutex.RLock()
defer fake.handleParticipantConnectRequestMutex.RUnlock()
fake.listNodesMutex.RLock()
defer fake.listNodesMutex.RUnlock()
fake.registerNodeMutex.RLock()
pkg/routing/routingfakes/fake_signal_client.go
-88
View File
@@ -7,7 +7,6 @@ import (
"github.com/livekit/livekit-server/pkg/routing"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/rpc"
)
type FakeSignalClient struct {
@@ -21,23 +20,6 @@ type FakeSignalClient struct {
activeCountReturnsOnCall map[int]struct {
result1 int
}
HandleParticipantConnectRequestStub func(context.Context, livekit.RoomName, livekit.ParticipantIdentity, livekit.NodeID, *rpc.RelaySignalv2ConnectRequest) (*rpc.RelaySignalv2ConnectResponse, error)
handleParticipantConnectRequestMutex sync.RWMutex
handleParticipantConnectRequestArgsForCall []struct {
arg1 context.Context
arg2 livekit.RoomName
arg3 livekit.ParticipantIdentity
arg4 livekit.NodeID
arg5 *rpc.RelaySignalv2ConnectRequest
}
handleParticipantConnectRequestReturns struct {
result1 *rpc.RelaySignalv2ConnectResponse
result2 error
}
handleParticipantConnectRequestReturnsOnCall map[int]struct {
result1 *rpc.RelaySignalv2ConnectResponse
result2 error
}
StartParticipantSignalStub func(context.Context, livekit.RoomName, routing.ParticipantInit, livekit.NodeID) (livekit.ConnectionID, routing.MessageSink, routing.MessageSource, error)
startParticipantSignalMutex sync.RWMutex
startParticipantSignalArgsForCall []struct {
@@ -115,74 +97,6 @@ func (fake *FakeSignalClient) ActiveCountReturnsOnCall(i int, result1 int) {
}{result1}
}
func (fake *FakeSignalClient) HandleParticipantConnectRequest(arg1 context.Context, arg2 livekit.RoomName, arg3 livekit.ParticipantIdentity, arg4 livekit.NodeID, arg5 *rpc.RelaySignalv2ConnectRequest) (*rpc.RelaySignalv2ConnectResponse, error) {
fake.handleParticipantConnectRequestMutex.Lock()
ret, specificReturn := fake.handleParticipantConnectRequestReturnsOnCall[len(fake.handleParticipantConnectRequestArgsForCall)]
fake.handleParticipantConnectRequestArgsForCall = append(fake.handleParticipantConnectRequestArgsForCall, struct {
arg1 context.Context
arg2 livekit.RoomName
arg3 livekit.ParticipantIdentity
arg4 livekit.NodeID
arg5 *rpc.RelaySignalv2ConnectRequest
}{arg1, arg2, arg3, arg4, arg5})
stub := fake.HandleParticipantConnectRequestStub
fakeReturns := fake.handleParticipantConnectRequestReturns
fake.recordInvocation("HandleParticipantConnectRequest", []interface{}{arg1, arg2, arg3, arg4, arg5})
fake.handleParticipantConnectRequestMutex.Unlock()
if stub != nil {
return stub(arg1, arg2, arg3, arg4, arg5)
}
if specificReturn {
return ret.result1, ret.result2
}
return fakeReturns.result1, fakeReturns.result2
}
func (fake *FakeSignalClient) HandleParticipantConnectRequestCallCount() int {
fake.handleParticipantConnectRequestMutex.RLock()
defer fake.handleParticipantConnectRequestMutex.RUnlock()
return len(fake.handleParticipantConnectRequestArgsForCall)
}
func (fake *FakeSignalClient) HandleParticipantConnectRequestCalls(stub func(context.Context, livekit.RoomName, livekit.ParticipantIdentity, livekit.NodeID, *rpc.RelaySignalv2ConnectRequest) (*rpc.RelaySignalv2ConnectResponse, error)) {
fake.handleParticipantConnectRequestMutex.Lock()
defer fake.handleParticipantConnectRequestMutex.Unlock()
fake.HandleParticipantConnectRequestStub = stub
}
func (fake *FakeSignalClient) HandleParticipantConnectRequestArgsForCall(i int) (context.Context, livekit.RoomName, livekit.ParticipantIdentity, livekit.NodeID, *rpc.RelaySignalv2ConnectRequest) {
fake.handleParticipantConnectRequestMutex.RLock()
defer fake.handleParticipantConnectRequestMutex.RUnlock()
argsForCall := fake.handleParticipantConnectRequestArgsForCall[i]
return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5
}
func (fake *FakeSignalClient) HandleParticipantConnectRequestReturns(result1 *rpc.RelaySignalv2ConnectResponse, result2 error) {
fake.handleParticipantConnectRequestMutex.Lock()
defer fake.handleParticipantConnectRequestMutex.Unlock()
fake.HandleParticipantConnectRequestStub = nil
fake.handleParticipantConnectRequestReturns = struct {
result1 *rpc.RelaySignalv2ConnectResponse
result2 error
}{result1, result2}
}
func (fake *FakeSignalClient) HandleParticipantConnectRequestReturnsOnCall(i int, result1 *rpc.RelaySignalv2ConnectResponse, result2 error) {
fake.handleParticipantConnectRequestMutex.Lock()
defer fake.handleParticipantConnectRequestMutex.Unlock()
fake.HandleParticipantConnectRequestStub = nil
if fake.handleParticipantConnectRequestReturnsOnCall == nil {
fake.handleParticipantConnectRequestReturnsOnCall = make(map[int]struct {
result1 *rpc.RelaySignalv2ConnectResponse
result2 error
})
}
fake.handleParticipantConnectRequestReturnsOnCall[i] = struct {
result1 *rpc.RelaySignalv2ConnectResponse
result2 error
}{result1, result2}
}
func (fake *FakeSignalClient) StartParticipantSignal(arg1 context.Context, arg2 livekit.RoomName, arg3 routing.ParticipantInit, arg4 livekit.NodeID) (livekit.ConnectionID, routing.MessageSink, routing.MessageSource, error) {
fake.startParticipantSignalMutex.Lock()
ret, specificReturn := fake.startParticipantSignalReturnsOnCall[len(fake.startParticipantSignalArgsForCall)]
@@ -261,8 +175,6 @@ func (fake *FakeSignalClient) Invocations() map[string][][]interface{} {
defer fake.invocationsMutex.RUnlock()
fake.activeCountMutex.RLock()
defer fake.activeCountMutex.RUnlock()
fake.handleParticipantConnectRequestMutex.RLock()
defer fake.handleParticipantConnectRequestMutex.RUnlock()
fake.startParticipantSignalMutex.RLock()
defer fake.startParticipantSignalMutex.RUnlock()
copiedInvocations := map[string][][]interface{}{}
pkg/routing/signal.go
+7 -46
View File
@@ -41,22 +41,13 @@ var ErrSignalMessageDropped = errors.New("signal message dropped")
type SignalClient interface {
ActiveCount() int
StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit, nodeID livekit.NodeID) (connectionID livekit.ConnectionID, reqSink MessageSink, resSource MessageSource, err error)
HandleParticipantConnectRequest(
ctx context.Context,
roomName livekit.RoomName,
participantIdentity livekit.ParticipantIdentity,
nodeID livekit.NodeID,
rscr *rpc.RelaySignalv2ConnectRequest,
) (*rpc.RelaySignalv2ConnectResponse, error)
}
type signalClient struct {
nodeID livekit.NodeID
config config.SignalRelayConfig
client rpc.TypedSignalClient
clientv2 rpc.TypedSignalv2Client
active atomic.Int32
nodeID livekit.NodeID
config config.SignalRelayConfig
client rpc.TypedSignalClient
active atomic.Int32
}
func NewSignalClient(nodeID livekit.NodeID, bus psrpc.MessageBus, config config.SignalRelayConfig) (SignalClient, error) {
@@ -70,20 +61,10 @@ func NewSignalClient(nodeID livekit.NodeID, bus psrpc.MessageBus, config config.
return nil, err
}
clientv2, err := rpc.NewTypedSignalv2Client(
nodeID,
bus,
middleware.WithClientMetrics(rpc.PSRPCMetricsObserver{}),
)
if err != nil {
return nil, err
}
return &signalClient{
nodeID: nodeID,
config: config,
client: client,
clientv2: clientv2,
nodeID: nodeID,
config: config,
client: client,
}, nil
}
@@ -161,26 +142,6 @@ func (r *signalClient) StartParticipantSignal(
return connectionID, sink, resChan, nil
}
func (r *signalClient) HandleParticipantConnectRequest(
ctx context.Context,
roomName livekit.RoomName,
participantIdentity livekit.ParticipantIdentity,
nodeID livekit.NodeID,
rscr *rpc.RelaySignalv2ConnectRequest,
) (*rpc.RelaySignalv2ConnectResponse, error) {
lgr := utils.GetLogger(ctx).WithValues(
"room", roomName,
"participant", participantIdentity,
"reqNodeID", nodeID,
// SIGNALLING-V2-TODO "connID", connectionID,
"connectRequest", logger.Proto(rscr),
)
lgr.Debugw("handling participant connect request")
return r.clientv2.RelaySignalv2Connect(ctx, nodeID, rscr)
}
// ------------------------------
type signalRequestMessageWriter struct{}
pkg/rtc/participant.go
+34 -70
View File
@@ -54,7 +54,6 @@ import (
"github.com/livekit/livekit-server/pkg/sfu"
"github.com/livekit/livekit-server/pkg/sfu/buffer"
"github.com/livekit/livekit-server/pkg/sfu/connectionquality"
"github.com/livekit/livekit-server/pkg/sfu/datachannel"
"github.com/livekit/livekit-server/pkg/sfu/mime"
"github.com/livekit/livekit-server/pkg/sfu/pacer"
"github.com/livekit/livekit-server/pkg/sfu/streamallocator"
@@ -201,7 +200,6 @@ type ParticipantParams struct {
DisableSenderReportPassThrough bool
MetricConfig metric.MetricConfig
UseOneShotSignallingMode bool
SynchronousLocalCandidatesMode bool
EnableMetrics bool
DataChannelMaxBufferedAmount uint64
DatachannelSlowThreshold int
@@ -1781,23 +1779,6 @@ func (h PublisherTransportHandler) OnDataMessageUnlabeled(data []byte) {
h.p.onReceivedDataMessageUnlabeled(data)
}
func (h PublisherTransportHandler) OnDataChannelOpenSignalling(dc *datachannel.DataChannelWriter[*webrtc.DataChannel]) {
sink := signalling.NewDataChannelMessageSink(signalling.DataChannelMessageSinkParams{
Logger: h.p.params.Logger,
DataChannel: dc,
})
h.p.signaller.SetResponseSink(sink)
}
func (h PublisherTransportHandler) OnDataChannelCloseSignalling(dc *datachannel.DataChannelWriter[*webrtc.DataChannel]) {
// SIGNALLING-V2-TODO: check that the closed data channel is actually the same as response sink
h.p.signaller.SetResponseSink(nil)
}
func (h PublisherTransportHandler) OnDataMessageSignalling(data []byte) {
h.p.signalHandler.HandleEncodedMessage(data)
}
func (h PublisherTransportHandler) OnDataSendError(err error) {
h.p.onDataSendError(err)
}
@@ -1841,33 +1822,17 @@ func (h PrimaryTransportHandler) OnFullyEstablished() {
}
func (p *ParticipantImpl) setupSignalling() {
// SIGNALLING-V2-TODO: do proper types to decide which signalling components to instantiate
if !p.params.SynchronousLocalCandidatesMode {
p.signalling = signalling.NewSignalling(signalling.SignallingParams{
Logger: p.params.Logger,
})
p.signalHandler = signalling.NewSignalHandler(signalling.SignalHandlerParams{
Logger: p.params.Logger,
Participant: p,
})
p.signaller = signalling.NewSignallerAsync(signalling.SignallerAsyncParams{
Logger: p.params.Logger,
Participant: p,
})
} else {
p.signalling = signalling.NewSignallingv2(signalling.Signallingv2Params{
Logger: p.params.Logger,
})
p.signalHandler = signalling.NewSignalHandlerv2(signalling.SignalHandlerv2Params{
Logger: p.params.Logger,
Participant: p,
Signalling: p.signalling,
})
p.signaller = signalling.NewSignallerv2Async(signalling.Signallerv2AsyncParams{
Logger: p.params.Logger,
Participant: p,
})
}
p.signalling = signalling.NewSignalling(signalling.SignallingParams{
Logger: p.params.Logger,
})
p.signalHandler = signalling.NewSignalHandler(signalling.SignalHandlerParams{
Logger: p.params.Logger,
Participant: p,
})
p.signaller = signalling.NewSignallerAsync(signalling.SignallerAsyncParams{
Logger: p.params.Logger,
Participant: p,
})
}
func (p *ParticipantImpl) setupTransportManager() error {
@@ -1889,30 +1854,29 @@ func (p *ParticipantImpl) setupTransportManager() error {
params := TransportManagerParams{
// primary connection does not change, canSubscribe can change if permission was updated
// after the participant has joined
SubscriberAsPrimary: subscriberAsPrimary,
Config: p.params.Config,
Twcc: p.twcc,
ProtocolVersion: p.params.ProtocolVersion,
CongestionControlConfig: p.params.CongestionControlConfig,
EnabledPublishCodecs: p.enabledPublishCodecs,
EnabledSubscribeCodecs: p.enabledSubscribeCodecs,
SimTracks: p.params.SimTracks,
ClientInfo: p.params.ClientInfo,
Migration: p.params.Migration,
AllowTCPFallback: p.params.AllowTCPFallback,
TCPFallbackRTTThreshold: p.params.TCPFallbackRTTThreshold,
AllowUDPUnstableFallback: p.params.AllowUDPUnstableFallback,
TURNSEnabled: p.params.TURNSEnabled,
AllowPlayoutDelay: p.params.PlayoutDelay.GetEnabled(),
DataChannelMaxBufferedAmount: p.params.DataChannelMaxBufferedAmount,
DatachannelSlowThreshold: p.params.DatachannelSlowThreshold,
Logger: p.params.Logger.WithComponent(sutils.ComponentTransport),
PublisherHandler: pth,
SubscriberHandler: sth,
DataChannelStats: p.dataChannelStats,
UseOneShotSignallingMode: p.params.UseOneShotSignallingMode,
SynchronousLocalCandidatesMode: p.params.SynchronousLocalCandidatesMode,
FireOnTrackBySdp: p.params.FireOnTrackBySdp,
SubscriberAsPrimary: subscriberAsPrimary,
Config: p.params.Config,
Twcc: p.twcc,
ProtocolVersion: p.params.ProtocolVersion,
CongestionControlConfig: p.params.CongestionControlConfig,
EnabledPublishCodecs: p.enabledPublishCodecs,
EnabledSubscribeCodecs: p.enabledSubscribeCodecs,
SimTracks: p.params.SimTracks,
ClientInfo: p.params.ClientInfo,
Migration: p.params.Migration,
AllowTCPFallback: p.params.AllowTCPFallback,
TCPFallbackRTTThreshold: p.params.TCPFallbackRTTThreshold,
AllowUDPUnstableFallback: p.params.AllowUDPUnstableFallback,
TURNSEnabled: p.params.TURNSEnabled,
AllowPlayoutDelay: p.params.PlayoutDelay.GetEnabled(),
DataChannelMaxBufferedAmount: p.params.DataChannelMaxBufferedAmount,
DatachannelSlowThreshold: p.params.DatachannelSlowThreshold,
Logger: p.params.Logger.WithComponent(sutils.ComponentTransport),
PublisherHandler: pth,
SubscriberHandler: sth,
DataChannelStats: p.dataChannelStats,
UseOneShotSignallingMode: p.params.UseOneShotSignallingMode,
FireOnTrackBySdp: p.params.FireOnTrackBySdp,
}
if p.params.SyncStreams && p.params.PlayoutDelay.GetEnabled() && p.params.ClientInfo.isFirefox() {
// we will disable playout delay for Firefox if the user is expecting
pkg/rtc/participant_signal.go
-36
View File
@@ -25,8 +25,6 @@ import (
"github.com/livekit/livekit-server/pkg/routing"
"github.com/livekit/livekit-server/pkg/rtc/types"
"google.golang.org/protobuf/proto"
)
func (p *ParticipantImpl) SetResponseSink(sink routing.MessageSink) {
@@ -332,37 +330,3 @@ func (p *ParticipantImpl) SendSubscriptionPermissionUpdate(publisherID livekit.P
}
return err
}
func (p *ParticipantImpl) SendConnectResponse(connectResponse *livekit.ConnectResponse) error {
// keep track of participant updates and versions
p.updateLock.Lock()
for _, op := range connectResponse.OtherParticipants {
p.updateCache.Add(livekit.ParticipantID(op.Sid), participantUpdateInfo{
identity: livekit.ParticipantIdentity(op.Identity),
version: op.Version,
state: op.State,
updatedAt: time.Now(),
})
}
p.updateLock.Unlock()
err := p.signaller.WriteMessage(p.signalling.SignalConnectResponse(connectResponse))
if err != nil {
return err
}
// update state after sending message, so that no participant updates could slip through before JoinResponse is sent
p.updateLock.Lock()
if p.State() == livekit.ParticipantInfo_JOINING {
p.updateState(livekit.ParticipantInfo_JOINED)
}
queuedUpdates := p.queuedUpdates
p.queuedUpdates = nil
p.updateLock.Unlock()
return p.SendParticipantUpdate(queuedUpdates)
}
func (p *ParticipantImpl) SignalPendingMessages() proto.Message {
return p.signalling.PendingMessages()
}
pkg/rtc/room.go
-218
View File
@@ -32,7 +32,6 @@ import (
"github.com/livekit/protocol/logger"
"github.com/livekit/protocol/observability/roomobs"
"github.com/livekit/protocol/rpc"
protosignalling "github.com/livekit/protocol/signalling"
"github.com/livekit/protocol/utils"
"github.com/livekit/protocol/utils/guid"
"github.com/livekit/psrpc"
@@ -616,199 +615,6 @@ func (r *Room) Join(
return nil
}
// SIGNALLING-V2-TODO: consolidate common parts between this function and Join()
func (r *Room) Joinv2(
participant types.LocalParticipant,
opts *ParticipantOptions,
iceServers []*livekit.ICEServer,
) error {
connectResponse, err := func() (*livekit.ConnectResponse, error) {
r.lock.Lock()
defer r.lock.Unlock()
if r.IsClosed() {
return nil, ErrRoomClosed
}
if r.participants[participant.Identity()] != nil {
return nil, ErrAlreadyJoined
}
if r.protoRoom.MaxParticipants > 0 && !participant.IsDependent() {
numParticipants := uint32(0)
for _, p := range r.participants {
if !p.IsDependent() {
numParticipants++
}
}
if numParticipants >= r.protoRoom.MaxParticipants {
return nil, ErrMaxParticipantsExceeded
}
}
if r.FirstJoinedAt() == 0 && !participant.IsDependent() {
r.joinedAt.Store(time.Now().Unix())
}
var onStateChangeMu sync.Mutex
participant.OnStateChange(func(p types.LocalParticipant) {
if r.onParticipantChanged != nil {
r.onParticipantChanged(p)
}
r.broadcastParticipantState(p, broadcastOptions{skipSource: true})
onStateChangeMu.Lock()
defer onStateChangeMu.Unlock()
if state := p.State(); state == livekit.ParticipantInfo_ACTIVE {
// subscribe participant to existing published tracks
r.subscribeToExistingTracks(p, false)
connectTime := time.Since(p.ConnectedAt())
meta := &livekit.AnalyticsClientMeta{
ClientConnectTime: uint32(connectTime.Milliseconds()),
}
infos := p.GetICEConnectionInfo()
var connectionType roomobs.ConnectionType
for _, info := range infos {
if info.Type != types.ICEConnectionTypeUnknown {
meta.ConnectionType = info.Type.String()
connectionType = info.Type.ReporterType()
break
}
}
r.telemetry.ParticipantActive(context.Background(),
r.ToProto(),
p.ToProto(),
meta,
false,
)
participant.GetReporter().Tx(func(tx roomobs.ParticipantSessionTx) {
tx.ReportClientConnectTime(uint16(connectTime.Milliseconds()))
tx.ReportConnectResult(roomobs.ConnectionResultSuccess)
tx.ReportConnectionType(connectionType)
})
fields := append(
connectionDetailsFields(infos),
"clientInfo", logger.Proto(sutils.ClientInfoWithoutAddress(p.GetClientInfo())),
)
p.GetLogger().Infow("participant active", fields...)
} else if state == livekit.ParticipantInfo_DISCONNECTED {
// remove participant from room
go r.RemoveParticipant(p.Identity(), p.ID(), p.CloseReason())
}
})
participant.OnSubscriberReady(func(p types.LocalParticipant) {
r.subscribeToExistingTracks(p, false)
})
// it's important to set this before connection, we don't want to miss out on any published tracks
participant.OnTrackPublished(r.onTrackPublished)
participant.OnTrackUpdated(r.onTrackUpdated)
participant.OnTrackUnpublished(r.onTrackUnpublished)
participant.OnParticipantUpdate(r.onParticipantUpdate)
participant.OnDataPacket(r.onDataPacket)
participant.OnDataMessage(r.onDataMessage)
participant.OnMetrics(r.onMetrics)
participant.OnSubscribeStatusChanged(func(publisherID livekit.ParticipantID, subscribed bool) {
if subscribed {
pub := r.GetParticipantByID(publisherID)
if pub != nil && pub.State() == livekit.ParticipantInfo_ACTIVE {
// when a participant subscribes to another participant,
// send speaker update if the subscribed to participant is active.
level, active := pub.GetAudioLevel()
if active {
_ = participant.SendSpeakerUpdate([]*livekit.SpeakerInfo{
{
Sid: string(pub.ID()),
Level: float32(level),
Active: active,
},
}, false)
}
if cq := pub.GetConnectionQuality(); cq != nil {
update := &livekit.ConnectionQualityUpdate{}
update.Updates = append(update.Updates, cq)
_ = participant.SendConnectionQualityUpdate(update)
}
}
} else {
// no longer subscribed to the publisher, clear speaker status
_ = participant.SendSpeakerUpdate([]*livekit.SpeakerInfo{
{
Sid: string(publisherID),
Level: 0,
Active: false,
},
}, true)
}
})
r.launchTargetAgents(maps.Values(r.agentDispatches), participant, livekit.JobType_JT_PARTICIPANT)
r.logger.Debugw(
"new participant joined",
"pID", participant.ID(),
"participant", participant.Identity(),
"clientInfo", logger.Proto(participant.GetClientInfo()),
"options", opts,
"numParticipants", len(r.participants),
)
if participant.IsRecorder() && !r.protoRoom.ActiveRecording {
r.protoRoom.ActiveRecording = true
r.protoProxy.MarkDirty(true)
} else {
r.protoProxy.MarkDirty(false)
}
r.participants[participant.Identity()] = participant
r.participantOpts[participant.Identity()] = opts
if r.onParticipantChanged != nil {
r.onParticipantChanged(participant)
}
time.AfterFunc(time.Minute, func() {
if !participant.Verify() {
r.RemoveParticipant(participant.Identity(), participant.ID(), types.ParticipantCloseReasonJoinTimeout)
}
})
connectResponse := r.createConnectResponseLocked(participant, iceServers)
participant.SetMigrateState(types.MigrateStateComplete)
return connectResponse, nil
}()
if err != nil {
return err
}
// SIGNALLING-V2-TODO
// 1. process published audio_tracks
// 2. process published video_tracks
// 3. HandleOffer and get answer (publisher)
r.subscribeToExistingTracks(participant, true)
offer, offerId, err := participant.GetOffer()
if err != nil {
participant.GetLogger().Warnw("could not get offer", err)
prometheus.ServiceOperationCounter.WithLabelValues("participant_join", "error", "get_subscriber_offer").Add(1)
return err
}
// SIGNALLING-V2-TODO: when moving to envelope response, do SendSdpOffer after sending connect response
connectResponse.SubscriberSdp = protosignalling.ToProtoSessionDescription(offer, offerId)
// for sync response, this does not actually send, only generates messageId and caches the message
if err := participant.SendConnectResponse(connectResponse); err != nil {
prometheus.ServiceOperationCounter.WithLabelValues("participant_join", "error", "send_response").Add(1)
return err
}
prometheus.ServiceOperationCounter.WithLabelValues("participant_join", "success", "").Add(1)
return nil
}
func (r *Room) ReplaceParticipantRequestSource(identity livekit.ParticipantIdentity, reqSource routing.MessageSource) {
r.lock.Lock()
if rs, ok := r.participantRequestSources[identity]; ok {
@@ -1275,30 +1081,6 @@ func (r *Room) createJoinResponseLocked(participant types.LocalParticipant, iceS
}
}
func (r *Room) createConnectResponseLocked(
participant types.LocalParticipant,
iceServers []*livekit.ICEServer,
) *livekit.ConnectResponse {
iceConfig := participant.GetICEConfig()
hasICEFallback := iceConfig.GetPreferencePublisher() != livekit.ICECandidateType_ICT_NONE || iceConfig.GetPreferenceSubscriber() != livekit.ICECandidateType_ICT_NONE
return &livekit.ConnectResponse{
Room: r.ToProto(),
Participant: participant.ToProto(),
OtherParticipants: GetOtherParticipantInfo(
participant,
false, // isMigratingIn
toParticipants(maps.Values(r.participants)),
false, // skipSubscriberBroadcast
),
IceServers: iceServers,
ClientConfiguration: participant.GetClientConfiguration(),
ServerInfo: r.serverInfo,
SifTrailer: r.trailer,
EnabledPublishCodecs: participant.GetEnabledPublishCodecs(),
FastPublish: participant.CanPublish() && !hasICEFallback,
}
}
// a ParticipantImpl in the room added a new track, subscribe other participants to it
func (r *Room) onTrackPublished(participant types.LocalParticipant, track types.MediaTrack) {
// publish participant update, since track state is changed
pkg/rtc/signalling/datachannel_messagesink.go
-77
View File
@@ -1,77 +0,0 @@
// Copyright 2023 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 signalling
import (
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
"github.com/pion/webrtc/v4"
"github.com/livekit/livekit-server/pkg/routing"
"github.com/livekit/livekit-server/pkg/sfu/datachannel"
"google.golang.org/protobuf/proto"
)
var _ routing.MessageSink = (*dataChannelMessageSink)(nil)
type DataChannelMessageSinkParams struct {
Logger logger.Logger
DataChannel *datachannel.DataChannelWriter[*webrtc.DataChannel]
}
type dataChannelMessageSink struct {
params DataChannelMessageSinkParams
}
func NewDataChannelMessageSink(params DataChannelMessageSinkParams) routing.MessageSink {
return &dataChannelMessageSink{
params: params,
}
}
func (d *dataChannelMessageSink) WriteMessage(msg proto.Message) error {
if msg == nil {
return nil
}
protoMsg, err := proto.Marshal(msg)
if err != nil {
d.params.Logger.Errorw("could not marshal message", err)
return err
}
if _, err := d.params.DataChannel.Write(protoMsg); err != nil {
// SIGNALLING-V2-TODO: filter out logging expected errors
d.params.Logger.Errorw("could not send message", err)
return err
}
return nil
}
func (d *dataChannelMessageSink) IsClosed() bool {
// SIGNALLING-V2-TODO
return false
}
func (d *dataChannelMessageSink) Close() {
// SIGNALLING-V2-TODO
}
func (d *dataChannelMessageSink) ConnectionID() livekit.ConnectionID {
// SIGNALLING-V2-TODO
return ""
}
pkg/rtc/signalling/interfaces.go
-9
View File
@@ -25,8 +25,6 @@ import (
type ParticipantSignalHandler interface {
HandleMessage(msg proto.Message) error
HandleEncodedMessage(data []byte) error
PruneStaleReassemblies()
}
type ParticipantSignaller interface {
@@ -59,11 +57,4 @@ type ParticipantSignalling interface {
SignalSubscribedQualityUpdate(subscribedQualityUpdate *livekit.SubscribedQualityUpdate) proto.Message
SignalSubscriptionResponse(subscriptionResponse *livekit.SubscriptionResponse) proto.Message
SignalSubscriptionPermissionUpdate(subscriptionPermissionUpdate *livekit.SubscriptionPermissionUpdate) proto.Message
AckMessageId(ackMessageId uint32)
SetLastProcessedRemoteMessageId(lastProcessedRemoteMessageId uint32)
PendingMessages() proto.Message
SignalConnectResponse(connectResponse *livekit.ConnectResponse) proto.Message
}
pkg/rtc/signalling/signalhandler.go
-10
View File
@@ -44,7 +44,6 @@ func NewSignalHandler(params SignalHandlerParams) ParticipantSignalHandler {
}
}
// SIGNALLING-V2-TODO: consolidate base message handling for messages common to different signalling versions
func (s *signalhandler) HandleMessage(msg proto.Message) error {
req, ok := msg.(*livekit.SignalRequest)
if !ok {
@@ -194,12 +193,3 @@ func (s *signalhandler) HandleMessage(msg proto.Message) error {
return nil
}
func (s *signalhandler) HandleEncodedMessage(data []byte) error {
signalRequest := &livekit.SignalRequest{}
if err := proto.Unmarshal(data, signalRequest); err != nil {
return err
}
return s.HandleMessage(signalRequest)
}
pkg/rtc/signalling/signalhandlerunimplemented.go
-6
View File
@@ -25,9 +25,3 @@ type signalhandlerUnimplemented struct{}
func (u *signalhandlerUnimplemented) HandleMessage(msg proto.Message) error {
return nil
}
func (u *signalhandlerUnimplemented) HandleEncodedMessage(data []byte) error {
return nil
}
func (u *signalhandlerUnimplemented) PruneStaleReassemblies() {}
pkg/rtc/signalling/signalhandlerv2.go
-150
View File
@@ -1,150 +0,0 @@
// Copyright 2023 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 signalling
import (
"fmt"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
protosignalling "github.com/livekit/protocol/signalling"
"google.golang.org/protobuf/proto"
"github.com/livekit/livekit-server/pkg/rtc/types"
"go.uber.org/atomic"
)
var _ ParticipantSignalHandler = (*signalhandlerv2)(nil)
type SignalHandlerv2Params struct {
Logger logger.Logger
Participant types.LocalParticipant
Signalling ParticipantSignalling
}
type signalhandlerv2 struct {
signalhandlerUnimplemented
params SignalHandlerv2Params
// SIGNALLING-V2-TODO: have to set this properly for `ConnectRequest` coming via sync HTTP path
lastProcessedRemoteMessageId atomic.Uint32
signalReassembler *protosignalling.SignalReassembler
}
func NewSignalHandlerv2(params SignalHandlerv2Params) ParticipantSignalHandler {
return &signalhandlerv2{
params: params,
signalReassembler: protosignalling.NewSignalReassembler(protosignalling.SignalReassemblerParams{
Logger: params.Logger,
}),
}
}
func (s *signalhandlerv2) HandleMessage(msg proto.Message) error {
req, ok := msg.(*livekit.Signalv2WireMessage)
if !ok {
s.params.Logger.Warnw(
"unknown message type", nil,
"messageType", fmt.Sprintf("%T", msg),
)
return ErrInvalidMessageType
}
// SIGNALLING-V2-TODO: check if this makes sense for data channel based signalling
s.params.Participant.UpdateLastSeenSignal()
switch msg := req.GetMessage().(type) {
case *livekit.Signalv2WireMessage_Envelope:
for _, clientMessage := range msg.Envelope.ClientMessages {
sequencer := clientMessage.GetSequencer()
if sequencer == nil || sequencer.MessageId == 0 {
s.params.Logger.Warnw(
"skipping message without sequencer", nil,
"messageType", fmt.Sprintf("%T", clientMessage),
)
continue
}
lprmi := s.lastProcessedRemoteMessageId.Load()
if sequencer.MessageId <= lprmi {
s.params.Logger.Infow(
"duplicate in message stream",
"last", lprmi,
"current", clientMessage.Sequencer.MessageId,
)
continue
}
// SIGNALLING-V2-TODO: ask for replay if there are gaps
if lprmi != 0 && sequencer.MessageId != lprmi+1 {
s.params.Logger.Infow(
"gap in message stream",
"last", lprmi,
"current", clientMessage.Sequencer.MessageId,
)
}
switch payload := clientMessage.GetMessage().(type) {
case *livekit.Signalv2ClientMessage_PublisherSdp:
s.params.Participant.HandleOffer(protosignalling.FromProtoSessionDescription(payload.PublisherSdp))
case *livekit.Signalv2ClientMessage_SubscriberSdp:
s.params.Participant.HandleAnswer(protosignalling.FromProtoSessionDescription(payload.SubscriberSdp))
case *livekit.Signalv2ClientMessage_Trickle:
candidateInit, err := protosignalling.FromProtoTrickle(payload.Trickle)
if err != nil {
s.params.Logger.Warnw("could not decode trickle", err)
return err
}
s.params.Participant.AddICECandidate(candidateInit, payload.Trickle.Target)
}
s.lastProcessedRemoteMessageId.Store(sequencer.MessageId)
s.params.Signalling.AckMessageId(sequencer.LastProcessedRemoteMessageId)
s.params.Signalling.SetLastProcessedRemoteMessageId(sequencer.MessageId)
}
case *livekit.Signalv2WireMessage_Fragment:
bytes := s.signalReassembler.Reassemble(msg.Fragment)
if len(bytes) != 0 {
wireMessage := &livekit.Signalv2WireMessage{}
err := proto.Unmarshal(bytes, wireMessage)
if err != nil {
s.params.Logger.Warnw("could not unmarshal re-assembled packet", err)
return err
}
s.HandleMessage(wireMessage)
}
}
return nil
}
func (s *signalhandlerv2) HandleEncodedMessage(data []byte) error {
wireMessage := &livekit.Signalv2WireMessage{}
if err := proto.Unmarshal(data, wireMessage); err != nil {
return err
}
return s.HandleMessage(wireMessage)
}
func (s *signalhandlerv2) PruneStaleReassemblies() {
s.signalReassembler.Prune()
}
pkg/rtc/signalling/signallerasync.go
-1
View File
@@ -83,7 +83,6 @@ func (s *signallerAsync) WriteMessage(msg proto.Message) error {
err := sink.WriteMessage(msg)
if err != nil {
// SIGNALLING-V2-TODO: check for data channel errors to treat as non-error
if utils.ErrorIsOneOf(err, psrpc.Canceled, routing.ErrChannelClosed) {
if typed, ok := msg.(*livekit.SignalResponse); ok {
s.params.Logger.Debugw(
pkg/rtc/signalling/signallerv2async.go
-173
View File
@@ -1,173 +0,0 @@
// Copyright 2023 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 signalling
import (
"fmt"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
protosignalling "github.com/livekit/protocol/signalling"
"github.com/livekit/protocol/utils"
"github.com/livekit/psrpc"
"github.com/livekit/livekit-server/pkg/routing"
"github.com/livekit/livekit-server/pkg/rtc/types"
"google.golang.org/protobuf/proto"
)
var _ ParticipantSignaller = (*signallerv2Async)(nil)
type Signallerv2AsyncParams struct {
Logger logger.Logger
Participant types.LocalParticipant
}
type signallerv2Async struct {
params Signallerv2AsyncParams
*signallerAsyncBase
signalSegmenter *protosignalling.SignalSegmenter
}
func NewSignallerv2Async(params Signallerv2AsyncParams) ParticipantSignaller {
return &signallerv2Async{
params: params,
signallerAsyncBase: newSignallerAsyncBase(signallerAsyncBaseParams{Logger: params.Logger}),
signalSegmenter: protosignalling.NewSignalSegmenter(protosignalling.SignalSegmenterParams{
Logger: params.Logger,
}),
}
}
// SIGNALLING-V2-TODO: need to lock write so that fragments do not get interrupted
func (s *signallerv2Async) WriteMessage(msg proto.Message) error {
if msg == nil {
return nil
}
if s.params.Participant.IsDisconnected() {
return nil
}
if !s.params.Participant.IsReady() {
if typed, ok := msg.(*livekit.Signalv2WireMessage); !ok {
s.params.Logger.Warnw(
"unknown message type", nil,
"messageType", fmt.Sprintf("%T", msg),
)
} else {
if !hasConnectResponse(typed) {
return nil
}
}
}
sink := s.GetResponseSink()
if sink == nil {
if typed, ok := msg.(*livekit.Signalv2WireMessage); ok {
s.params.Logger.Debugw(
"could not send message to participant",
"messageType", fmt.Sprintf("%T", typed.Message),
)
}
return nil
}
// SIGNALLING-V2-TODO: avoid double marshalling,
// have to marshal once to get size of serialised packet and decide if it needs fragmentation,
// should used the marshaled bytes if fragmentation is not needed
var fragments []*livekit.Fragment
marshaled, err := proto.Marshal(msg)
if err != nil {
if typed, ok := msg.(*livekit.Signalv2WireMessage); ok {
s.params.Logger.Warnw(
"could not send message to participant", err,
"messageType", fmt.Sprintf("%T", typed.Message),
)
}
} else {
fragments = s.signalSegmenter.Segment(marshaled)
}
sendMsg := func(m proto.Message) error {
if err := sink.WriteMessage(m); err != nil {
// SIGNALLING-V2-TODO: check for data channel errors to treat as debug too
if utils.ErrorIsOneOf(err, psrpc.Canceled, routing.ErrChannelClosed) {
if typed, ok := m.(*livekit.Signalv2WireMessage); ok {
s.params.Logger.Debugw(
"could not send message to participant",
"error", err,
"messageType", fmt.Sprintf("%T", typed.Message),
)
}
return nil
} else {
if typed, ok := m.(*livekit.Signalv2WireMessage); ok {
s.params.Logger.Warnw(
"could not send message to participant", err,
"messageType", fmt.Sprintf("%T", typed.Message),
)
}
return err
}
}
return nil
}
if len(fragments) != 0 {
for _, fragment := range fragments {
wireMessage := &livekit.Signalv2WireMessage{
Message: &livekit.Signalv2WireMessage_Fragment{
Fragment: fragment,
},
}
if err := sendMsg(wireMessage); err != nil {
return err
}
}
} else {
if err := sendMsg(msg); err != nil {
return err
}
}
return nil
}
// ----------------------------
func hasConnectResponse(wireMessage *livekit.Signalv2WireMessage) bool {
switch msg := wireMessage.GetMessage().(type) {
case *livekit.Signalv2WireMessage_Envelope:
for _, innerMsg := range msg.Envelope.GetServerMessages() {
switch innerMsg.GetMessage().(type) {
case *livekit.Signalv2ServerMessage_ConnectResponse:
return true
default:
return false // first message should be `ConnectResponse`
}
}
default:
// SIGNALLING-V2-TODO: handle ConnectResponse getting fragmented.
return false
}
return false
}
pkg/rtc/signalling/signallerv2hybrid.go
-38
View File
@@ -1,38 +0,0 @@
// Copyright 2023 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 signalling
import (
"github.com/livekit/protocol/logger"
)
var _ ParticipantSignaller = (*signallerv2Hybrid)(nil)
type Signallerv2HybridParams struct {
Logger logger.Logger
}
type signallerv2Hybrid struct {
params Signallerv2HybridParams
*signallerv2Async
}
func NewSignallerv2Hybrid(params Signallerv2HybridParams) ParticipantSignaller {
return &signallerv2Hybrid{
params: params,
signallerv2Async: NewSignallerv2Async(Signallerv2AsyncParams{Logger: params.Logger}).(*signallerv2Async),
}
}
pkg/rtc/signalling/signallingunimplemented.go
-13
View File
@@ -107,16 +107,3 @@ func (u *signallingUnimplemented) SignalSubscriptionResponse(subscriptionRespons
func (u *signallingUnimplemented) SignalSubscriptionPermissionUpdate(subscriptionPermissionUpdate *livekit.SubscriptionPermissionUpdate) proto.Message {
return nil
}
func (u *signallingUnimplemented) AckMessageId(ackMessageId uint32) {}
func (u *signallingUnimplemented) SetLastProcessedRemoteMessageId(lastProcessedRemoteMessageId uint32) {
}
func (u *signallingUnimplemented) PendingMessages() proto.Message {
return nil
}
func (u *signallingUnimplemented) SignalConnectResponse(connectResponse *livekit.ConnectResponse) proto.Message {
return nil
}
pkg/rtc/signalling/signallingv2.go
-136
View File
@@ -1,136 +0,0 @@
// Copyright 2023 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 signalling
import (
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
protosignalling "github.com/livekit/protocol/signalling"
"google.golang.org/protobuf/proto"
)
var _ ParticipantSignalling = (*signallingv2)(nil)
type Signallingv2Params struct {
Logger logger.Logger
}
type signallingv2 struct {
signallingUnimplemented
params Signallingv2Params
signalCache *protosignalling.Signalv2ServerMessageCache
}
func NewSignallingv2(params Signallingv2Params) ParticipantSignalling {
return &signallingv2{
params: params,
signalCache: protosignalling.NewSignalv2ServerMessageCache(protosignalling.SignalCacheParams{
Logger: params.Logger,
}),
}
}
func (s *signallingv2) AckMessageId(ackMessageId uint32) {
s.signalCache.Clear(ackMessageId)
}
func (s *signallingv2) SetLastProcessedRemoteMessageId(lastProcessedRemoteMessageId uint32) {
s.signalCache.SetLastProcessedRemoteMessageId(lastProcessedRemoteMessageId)
}
func (s *signallingv2) PendingMessages() proto.Message {
serverMessages := s.signalCache.GetFromFront()
if len(serverMessages) == 0 {
return nil
}
return &livekit.Signalv2WireMessage{
Message: &livekit.Signalv2WireMessage_Envelope{
Envelope: &livekit.Envelope{
ServerMessages: serverMessages,
},
},
}
}
func (s *signallingv2) SignalConnectResponse(connectResponse *livekit.ConnectResponse) proto.Message {
serverMessage := &livekit.Signalv2ServerMessage{
Message: &livekit.Signalv2ServerMessage_ConnectResponse{
ConnectResponse: connectResponse,
},
}
return s.cacheAndReturnEnvelope(serverMessage)
}
func (s *signallingv2) SignalSdpOffer(offer *livekit.SessionDescription) proto.Message {
serverMessage := &livekit.Signalv2ServerMessage{
Message: &livekit.Signalv2ServerMessage_SubscriberSdp{
SubscriberSdp: offer,
},
}
return s.cacheAndReturnEnvelope(serverMessage)
}
func (s *signallingv2) SignalSdpAnswer(answer *livekit.SessionDescription) proto.Message {
serverMessage := &livekit.Signalv2ServerMessage{
Message: &livekit.Signalv2ServerMessage_PublisherSdp{
PublisherSdp: answer,
},
}
return s.cacheAndReturnEnvelope(serverMessage)
}
func (s *signallingv2) SignalRoomUpdate(room *livekit.Room) proto.Message {
serverMessage := &livekit.Signalv2ServerMessage{
Message: &livekit.Signalv2ServerMessage_RoomUpdate{
RoomUpdate: &livekit.RoomUpdate{
Room: room,
},
},
}
return s.cacheAndReturnEnvelope(serverMessage)
}
func (s *signallingv2) SignalParticipantUpdate(participants []*livekit.ParticipantInfo) proto.Message {
if len(participants) == 0 {
return nil
}
serverMessage := &livekit.Signalv2ServerMessage{
Message: &livekit.Signalv2ServerMessage_ParticipantUpdate{
ParticipantUpdate: &livekit.ParticipantUpdate{
Participants: participants,
},
},
}
return s.cacheAndReturnEnvelope(serverMessage)
}
func (s *signallingv2) cacheAndReturnEnvelope(sm *livekit.Signalv2ServerMessage) proto.Message {
sm = s.signalCache.Add(sm)
if sm == nil {
return nil
}
return &livekit.Signalv2WireMessage{
Message: &livekit.Signalv2WireMessage_Envelope{
Envelope: &livekit.Envelope{
ServerMessages: []*livekit.Signalv2ServerMessage{sm},
},
},
}
}
pkg/rtc/transport.go
+21 -73
View File
@@ -266,25 +266,24 @@ type PCTransport struct {
}
type TransportParams struct {
Handler transport.Handler
ProtocolVersion types.ProtocolVersion
Config *WebRTCConfig
Twcc *lktwcc.Responder
DirectionConfig DirectionConfig
CongestionControlConfig config.CongestionControlConfig
EnabledCodecs []*livekit.Codec
Logger logger.Logger
Transport livekit.SignalTarget
SimTracks map[uint32]SimulcastTrackInfo
ClientInfo ClientInfo
IsOfferer bool
IsSendSide bool
AllowPlayoutDelay bool
UseOneShotSignallingMode bool
SynchronousLocalCandidatesMode bool
FireOnTrackBySdp bool
DataChannelMaxBufferedAmount uint64
DatachannelSlowThreshold int
Handler transport.Handler
ProtocolVersion types.ProtocolVersion
Config *WebRTCConfig
Twcc *lktwcc.Responder
DirectionConfig DirectionConfig
CongestionControlConfig config.CongestionControlConfig
EnabledCodecs []*livekit.Codec
Logger logger.Logger
Transport livekit.SignalTarget
SimTracks map[uint32]SimulcastTrackInfo
ClientInfo ClientInfo
IsOfferer bool
IsSendSide bool
AllowPlayoutDelay bool
UseOneShotSignallingMode bool
FireOnTrackBySdp bool
DataChannelMaxBufferedAmount uint64
DatachannelSlowThreshold int
// for development test
DatachannelMaxReceiverBufferSize int
@@ -555,8 +554,7 @@ func (t *PCTransport) createPeerConnection() (cc.BandwidthEstimator, error) {
}
t.pc = pc
// SIGNALLING-V2-TODO: have to support both sync and async candidates, so has to be a check at function level
if !t.params.UseOneShotSignallingMode /* SIGNALLING-V2-TODO && !t.params.SynchronousLocalCandidatesMode */ {
if !t.params.UseOneShotSignallingMode {
// one shot signalling mode gathers all candidates and sends in answer
t.pc.OnICEGatheringStateChange(t.onICEGatheringStateChange)
t.pc.OnICECandidate(t.onICECandidateTrickle)
@@ -1330,10 +1328,8 @@ func (t *PCTransport) clearConnTimer() {
}
}
// SIGNALLING-V2-TODO: this needs both sync and async support when not in one shot mode,
// cannot use the state `SynchronousLocalCandidatesMode`, needs a flag at function level
func (t *PCTransport) HandleRemoteDescription(sd webrtc.SessionDescription, remoteId uint32) error {
if t.params.UseOneShotSignallingMode /* SIGNALLING-V2-TODO || t.params.SynchronousLocalCandidatesMode */ {
if t.params.UseOneShotSignallingMode {
if sd.Type == webrtc.SDPTypeOffer {
remoteOfferId := t.remoteOfferId.Load()
if remoteOfferId != 0 && remoteOfferId != t.localAnswerId.Load() {
@@ -1352,7 +1348,6 @@ func (t *PCTransport) HandleRemoteDescription(sd webrtc.SessionDescription, remo
t.remoteAnswerId.Store(remoteId)
}
// SIGNALLING-V2-TODO: need to support filtering candidates for transport fallback
// add remote candidates to ICE connection details
parsed, err := sd.Unmarshal()
if err == nil {
@@ -1400,10 +1395,8 @@ func (t *PCTransport) HandleRemoteDescription(sd webrtc.SessionDescription, remo
return nil
}
// SIGNALLING-V2-TODO: use a flag at function level for sync vs async rather
// then state `SynchronousLocalCandidatesMode`
func (t *PCTransport) GetAnswer() (webrtc.SessionDescription, uint32, error) {
if !t.params.UseOneShotSignallingMode && !t.params.SynchronousLocalCandidatesMode {
if !t.params.UseOneShotSignallingMode {
return webrtc.SessionDescription{}, 0, ErrNotSynchronousLocalCandidatesMode
}
@@ -1453,51 +1446,6 @@ func (t *PCTransport) GetAnswer() (webrtc.SessionDescription, uint32, error) {
return *cld, answerId, nil
}
// SIGNALLING-V2-TODO: use a flag at function level for sync vs async rather
// then state `SynchronousLocalCandidatesMode`
func (t *PCTransport) GetOffer() (webrtc.SessionDescription, uint32, error) {
if !t.params.SynchronousLocalCandidatesMode {
return webrtc.SessionDescription{}, 0, ErrNotSynchronousLocalCandidatesMode
}
offer, err := t.pc.CreateOffer(nil)
if err != nil {
return webrtc.SessionDescription{}, 0, err
}
if err = t.pc.SetLocalDescription(offer); err != nil {
return webrtc.SessionDescription{}, 0, err
}
// wait for gathering to complete to include all candidates in the answer
<-webrtc.GatheringCompletePromise(t.pc)
pld := t.pc.PendingLocalDescription()
// add local candidates to ICE connection details
parsed, err := pld.Unmarshal()
if err == nil {
addLocalICECandidates := func(attrs []sdp.Attribute) {
for _, a := range attrs {
if a.IsICECandidate() {
c, err := ice.UnmarshalCandidate(a.Value)
if err != nil {
continue
}
t.connectionDetails.AddLocalICECandidate(c, false, false)
}
}
}
addLocalICECandidates(parsed.Attributes)
for _, m := range parsed.MediaDescriptions {
addLocalICECandidates(m.Attributes)
}
}
return *pld, t.localOfferId.Inc(), nil
}
func (t *PCTransport) GetICESessionUfrag() (string, error) {
cld := t.pc.CurrentLocalDescription()
if cld == nil {
pkg/rtc/transportmanager.go
+51 -58
View File
@@ -83,30 +83,29 @@ func (h TransportManagerPublisherTransportHandler) OnAnswer(sd webrtc.SessionDes
// -------------------------------
type TransportManagerParams struct {
SubscriberAsPrimary bool
Config *WebRTCConfig
Twcc *twcc.Responder
ProtocolVersion types.ProtocolVersion
CongestionControlConfig config.CongestionControlConfig
EnabledSubscribeCodecs []*livekit.Codec
EnabledPublishCodecs []*livekit.Codec
SimTracks map[uint32]SimulcastTrackInfo
ClientInfo ClientInfo
Migration bool
AllowTCPFallback bool
TCPFallbackRTTThreshold int
AllowUDPUnstableFallback bool
TURNSEnabled bool
AllowPlayoutDelay bool
DataChannelMaxBufferedAmount uint64
DatachannelSlowThreshold int
Logger logger.Logger
PublisherHandler transport.Handler
SubscriberHandler transport.Handler
DataChannelStats *telemetry.BytesTrackStats
UseOneShotSignallingMode bool
SynchronousLocalCandidatesMode bool
FireOnTrackBySdp bool
SubscriberAsPrimary bool
Config *WebRTCConfig
Twcc *twcc.Responder
ProtocolVersion types.ProtocolVersion
CongestionControlConfig config.CongestionControlConfig
EnabledSubscribeCodecs []*livekit.Codec
EnabledPublishCodecs []*livekit.Codec
SimTracks map[uint32]SimulcastTrackInfo
ClientInfo ClientInfo
Migration bool
AllowTCPFallback bool
TCPFallbackRTTThreshold int
AllowUDPUnstableFallback bool
TURNSEnabled bool
AllowPlayoutDelay bool
DataChannelMaxBufferedAmount uint64
DatachannelSlowThreshold int
Logger logger.Logger
PublisherHandler transport.Handler
SubscriberHandler transport.Handler
DataChannelStats *telemetry.BytesTrackStats
UseOneShotSignallingMode bool
FireOnTrackBySdp bool
}
type TransportManager struct {
@@ -151,22 +150,21 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro
lgr := LoggerWithPCTarget(params.Logger, livekit.SignalTarget_PUBLISHER)
publisher, err := NewPCTransport(TransportParams{
ProtocolVersion: params.ProtocolVersion,
Config: params.Config,
Twcc: params.Twcc,
DirectionConfig: params.Config.Publisher,
CongestionControlConfig: params.CongestionControlConfig,
EnabledCodecs: params.EnabledPublishCodecs,
Logger: lgr,
SimTracks: params.SimTracks,
ClientInfo: params.ClientInfo,
Transport: livekit.SignalTarget_PUBLISHER,
Handler: TransportManagerPublisherTransportHandler{TransportManagerTransportHandler{params.PublisherHandler, t, lgr}},
UseOneShotSignallingMode: params.UseOneShotSignallingMode,
SynchronousLocalCandidatesMode: params.SynchronousLocalCandidatesMode,
DataChannelMaxBufferedAmount: params.DataChannelMaxBufferedAmount,
DatachannelSlowThreshold: params.DatachannelSlowThreshold,
FireOnTrackBySdp: params.FireOnTrackBySdp,
ProtocolVersion: params.ProtocolVersion,
Config: params.Config,
Twcc: params.Twcc,
DirectionConfig: params.Config.Publisher,
CongestionControlConfig: params.CongestionControlConfig,
EnabledCodecs: params.EnabledPublishCodecs,
Logger: lgr,
SimTracks: params.SimTracks,
ClientInfo: params.ClientInfo,
Transport: livekit.SignalTarget_PUBLISHER,
Handler: TransportManagerPublisherTransportHandler{TransportManagerTransportHandler{params.PublisherHandler, t, lgr}},
UseOneShotSignallingMode: params.UseOneShotSignallingMode,
DataChannelMaxBufferedAmount: params.DataChannelMaxBufferedAmount,
DatachannelSlowThreshold: params.DatachannelSlowThreshold,
FireOnTrackBySdp: params.FireOnTrackBySdp,
})
if err != nil {
return nil, err
@@ -175,20 +173,19 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro
lgr = LoggerWithPCTarget(params.Logger, livekit.SignalTarget_SUBSCRIBER)
subscriber, err := NewPCTransport(TransportParams{
ProtocolVersion: params.ProtocolVersion,
Config: params.Config,
DirectionConfig: params.Config.Subscriber,
CongestionControlConfig: params.CongestionControlConfig,
EnabledCodecs: params.EnabledSubscribeCodecs,
Logger: lgr,
ClientInfo: params.ClientInfo,
IsOfferer: true,
IsSendSide: true,
AllowPlayoutDelay: params.AllowPlayoutDelay,
DatachannelSlowThreshold: params.DatachannelSlowThreshold,
Transport: livekit.SignalTarget_SUBSCRIBER,
Handler: TransportManagerTransportHandler{params.SubscriberHandler, t, lgr},
SynchronousLocalCandidatesMode: params.SynchronousLocalCandidatesMode,
ProtocolVersion: params.ProtocolVersion,
Config: params.Config,
DirectionConfig: params.Config.Subscriber,
CongestionControlConfig: params.CongestionControlConfig,
EnabledCodecs: params.EnabledSubscribeCodecs,
Logger: lgr,
ClientInfo: params.ClientInfo,
IsOfferer: true,
IsSendSide: true,
AllowPlayoutDelay: params.AllowPlayoutDelay,
DatachannelSlowThreshold: params.DatachannelSlowThreshold,
Transport: livekit.SignalTarget_SUBSCRIBER,
Handler: TransportManagerTransportHandler{params.SubscriberHandler, t, lgr},
})
if err != nil {
return nil, err
@@ -496,10 +493,6 @@ func (t *TransportManager) HandleAnswer(answer webrtc.SessionDescription, answer
t.subscriber.HandleRemoteDescription(answer, answerId)
}
func (t *TransportManager) GetOffer() (webrtc.SessionDescription, uint32, error) {
return t.subscriber.GetOffer()
}
// AddICECandidate adds candidates for remote peer
func (t *TransportManager) AddICECandidate(candidate webrtc.ICECandidateInit, target livekit.SignalTarget) {
switch target {
pkg/rtc/types/interfaces.go
-3
View File
@@ -411,7 +411,6 @@ type LocalParticipant interface {
SetTrackMuted(trackID livekit.TrackID, muted bool, fromAdmin bool) *livekit.TrackInfo
HandleAnswer(sdp webrtc.SessionDescription, answerId uint32)
GetOffer() (webrtc.SessionDescription, uint32, error)
Negotiate(force bool)
ICERestart(iceConfig *livekit.ICEConfig)
AddTrackLocal(trackLocal webrtc.TrackLocal, params AddTrackParams) (*webrtc.RTPSender, *webrtc.RTPTransceiver, error)
@@ -454,8 +453,6 @@ type LocalParticipant interface {
HandleReconnectAndSendResponse(reconnectReason livekit.ReconnectReason, reconnectResponse *livekit.ReconnectResponse) error
IssueFullReconnect(reason ParticipantCloseReason)
SendRoomMovedResponse(moved *livekit.RoomMovedResponse) error
SendConnectResponse(connectResponse *livekit.ConnectResponse) error
SignalPendingMessages() proto.Message
// callbacks
OnStateChange(func(p LocalParticipant))
pkg/rtc/types/typesfakes/fake_local_participant.go
-214
View File
@@ -373,20 +373,6 @@ type FakeLocalParticipant struct {
getLoggerResolverReturnsOnCall map[int]struct {
result1 logger.DeferredFieldResolver
}
GetOfferStub func() (webrtc.SessionDescription, uint32, error)
getOfferMutex sync.RWMutex
getOfferArgsForCall []struct {
}
getOfferReturns struct {
result1 webrtc.SessionDescription
result2 uint32
result3 error
}
getOfferReturnsOnCall map[int]struct {
result1 webrtc.SessionDescription
result2 uint32
result3 error
}
GetPacerStub func() pacer.Pacer
getPacerMutex sync.RWMutex
getPacerArgsForCall []struct {
@@ -986,17 +972,6 @@ type FakeLocalParticipant struct {
removeTrackLocalReturnsOnCall map[int]struct {
result1 error
}
SendConnectResponseStub func(*livekit.ConnectResponse) error
sendConnectResponseMutex sync.RWMutex
sendConnectResponseArgsForCall []struct {
arg1 *livekit.ConnectResponse
}
sendConnectResponseReturns struct {
result1 error
}
sendConnectResponseReturnsOnCall map[int]struct {
result1 error
}
SendConnectionQualityUpdateStub func(*livekit.ConnectionQualityUpdate) error
sendConnectionQualityUpdateMutex sync.RWMutex
sendConnectionQualityUpdateArgsForCall []struct {
@@ -1204,16 +1179,6 @@ type FakeLocalParticipant struct {
setTrackMutedReturnsOnCall map[int]struct {
result1 *livekit.TrackInfo
}
SignalPendingMessagesStub func() proto.Message
signalPendingMessagesMutex sync.RWMutex
signalPendingMessagesArgsForCall []struct {
}
signalPendingMessagesReturns struct {
result1 proto.Message
}
signalPendingMessagesReturnsOnCall map[int]struct {
result1 proto.Message
}
StateStub func() livekit.ParticipantInfo_State
stateMutex sync.RWMutex
stateArgsForCall []struct {
@@ -3274,65 +3239,6 @@ func (fake *FakeLocalParticipant) GetLoggerResolverReturnsOnCall(i int, result1
}{result1}
}
func (fake *FakeLocalParticipant) GetOffer() (webrtc.SessionDescription, uint32, error) {
fake.getOfferMutex.Lock()
ret, specificReturn := fake.getOfferReturnsOnCall[len(fake.getOfferArgsForCall)]
fake.getOfferArgsForCall = append(fake.getOfferArgsForCall, struct {
}{})
stub := fake.GetOfferStub
fakeReturns := fake.getOfferReturns
fake.recordInvocation("GetOffer", []interface{}{})
fake.getOfferMutex.Unlock()
if stub != nil {
return stub()
}
if specificReturn {
return ret.result1, ret.result2, ret.result3
}
return fakeReturns.result1, fakeReturns.result2, fakeReturns.result3
}
func (fake *FakeLocalParticipant) GetOfferCallCount() int {
fake.getOfferMutex.RLock()
defer fake.getOfferMutex.RUnlock()
return len(fake.getOfferArgsForCall)
}
func (fake *FakeLocalParticipant) GetOfferCalls(stub func() (webrtc.SessionDescription, uint32, error)) {
fake.getOfferMutex.Lock()
defer fake.getOfferMutex.Unlock()
fake.GetOfferStub = stub
}
func (fake *FakeLocalParticipant) GetOfferReturns(result1 webrtc.SessionDescription, result2 uint32, result3 error) {
fake.getOfferMutex.Lock()
defer fake.getOfferMutex.Unlock()
fake.GetOfferStub = nil
fake.getOfferReturns = struct {
result1 webrtc.SessionDescription
result2 uint32
result3 error
}{result1, result2, result3}
}
func (fake *FakeLocalParticipant) GetOfferReturnsOnCall(i int, result1 webrtc.SessionDescription, result2 uint32, result3 error) {
fake.getOfferMutex.Lock()
defer fake.getOfferMutex.Unlock()
fake.GetOfferStub = nil
if fake.getOfferReturnsOnCall == nil {
fake.getOfferReturnsOnCall = make(map[int]struct {
result1 webrtc.SessionDescription
result2 uint32
result3 error
})
}
fake.getOfferReturnsOnCall[i] = struct {
result1 webrtc.SessionDescription
result2 uint32
result3 error
}{result1, result2, result3}
}
func (fake *FakeLocalParticipant) GetPacer() pacer.Pacer {
fake.getPacerMutex.Lock()
ret, specificReturn := fake.getPacerReturnsOnCall[len(fake.getPacerArgsForCall)]
@@ -6658,67 +6564,6 @@ func (fake *FakeLocalParticipant) RemoveTrackLocalReturnsOnCall(i int, result1 e
}{result1}
}
func (fake *FakeLocalParticipant) SendConnectResponse(arg1 *livekit.ConnectResponse) error {
fake.sendConnectResponseMutex.Lock()
ret, specificReturn := fake.sendConnectResponseReturnsOnCall[len(fake.sendConnectResponseArgsForCall)]
fake.sendConnectResponseArgsForCall = append(fake.sendConnectResponseArgsForCall, struct {
arg1 *livekit.ConnectResponse
}{arg1})
stub := fake.SendConnectResponseStub
fakeReturns := fake.sendConnectResponseReturns
fake.recordInvocation("SendConnectResponse", []interface{}{arg1})
fake.sendConnectResponseMutex.Unlock()
if stub != nil {
return stub(arg1)
}
if specificReturn {
return ret.result1
}
return fakeReturns.result1
}
func (fake *FakeLocalParticipant) SendConnectResponseCallCount() int {
fake.sendConnectResponseMutex.RLock()
defer fake.sendConnectResponseMutex.RUnlock()
return len(fake.sendConnectResponseArgsForCall)
}
func (fake *FakeLocalParticipant) SendConnectResponseCalls(stub func(*livekit.ConnectResponse) error) {
fake.sendConnectResponseMutex.Lock()
defer fake.sendConnectResponseMutex.Unlock()
fake.SendConnectResponseStub = stub
}
func (fake *FakeLocalParticipant) SendConnectResponseArgsForCall(i int) *livekit.ConnectResponse {
fake.sendConnectResponseMutex.RLock()
defer fake.sendConnectResponseMutex.RUnlock()
argsForCall := fake.sendConnectResponseArgsForCall[i]
return argsForCall.arg1
}
func (fake *FakeLocalParticipant) SendConnectResponseReturns(result1 error) {
fake.sendConnectResponseMutex.Lock()
defer fake.sendConnectResponseMutex.Unlock()
fake.SendConnectResponseStub = nil
fake.sendConnectResponseReturns = struct {
result1 error
}{result1}
}
func (fake *FakeLocalParticipant) SendConnectResponseReturnsOnCall(i int, result1 error) {
fake.sendConnectResponseMutex.Lock()
defer fake.sendConnectResponseMutex.Unlock()
fake.SendConnectResponseStub = nil
if fake.sendConnectResponseReturnsOnCall == nil {
fake.sendConnectResponseReturnsOnCall = make(map[int]struct {
result1 error
})
}
fake.sendConnectResponseReturnsOnCall[i] = struct {
result1 error
}{result1}
}
func (fake *FakeLocalParticipant) SendConnectionQualityUpdate(arg1 *livekit.ConnectionQualityUpdate) error {
fake.sendConnectionQualityUpdateMutex.Lock()
ret, specificReturn := fake.sendConnectionQualityUpdateReturnsOnCall[len(fake.sendConnectionQualityUpdateArgsForCall)]
@@ -7881,59 +7726,6 @@ func (fake *FakeLocalParticipant) SetTrackMutedReturnsOnCall(i int, result1 *liv
}{result1}
}
func (fake *FakeLocalParticipant) SignalPendingMessages() proto.Message {
fake.signalPendingMessagesMutex.Lock()
ret, specificReturn := fake.signalPendingMessagesReturnsOnCall[len(fake.signalPendingMessagesArgsForCall)]
fake.signalPendingMessagesArgsForCall = append(fake.signalPendingMessagesArgsForCall, struct {
}{})
stub := fake.SignalPendingMessagesStub
fakeReturns := fake.signalPendingMessagesReturns
fake.recordInvocation("SignalPendingMessages", []interface{}{})
fake.signalPendingMessagesMutex.Unlock()
if stub != nil {
return stub()
}
if specificReturn {
return ret.result1
}
return fakeReturns.result1
}
func (fake *FakeLocalParticipant) SignalPendingMessagesCallCount() int {
fake.signalPendingMessagesMutex.RLock()
defer fake.signalPendingMessagesMutex.RUnlock()
return len(fake.signalPendingMessagesArgsForCall)
}
func (fake *FakeLocalParticipant) SignalPendingMessagesCalls(stub func() proto.Message) {
fake.signalPendingMessagesMutex.Lock()
defer fake.signalPendingMessagesMutex.Unlock()
fake.SignalPendingMessagesStub = stub
}
func (fake *FakeLocalParticipant) SignalPendingMessagesReturns(result1 proto.Message) {
fake.signalPendingMessagesMutex.Lock()
defer fake.signalPendingMessagesMutex.Unlock()
fake.SignalPendingMessagesStub = nil
fake.signalPendingMessagesReturns = struct {
result1 proto.Message
}{result1}
}
func (fake *FakeLocalParticipant) SignalPendingMessagesReturnsOnCall(i int, result1 proto.Message) {
fake.signalPendingMessagesMutex.Lock()
defer fake.signalPendingMessagesMutex.Unlock()
fake.SignalPendingMessagesStub = nil
if fake.signalPendingMessagesReturnsOnCall == nil {
fake.signalPendingMessagesReturnsOnCall = make(map[int]struct {
result1 proto.Message
})
}
fake.signalPendingMessagesReturnsOnCall[i] = struct {
result1 proto.Message
}{result1}
}
func (fake *FakeLocalParticipant) State() livekit.ParticipantInfo_State {
fake.stateMutex.Lock()
ret, specificReturn := fake.stateReturnsOnCall[len(fake.stateArgsForCall)]
@@ -9341,8 +9133,6 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
defer fake.getLoggerMutex.RUnlock()
fake.getLoggerResolverMutex.RLock()
defer fake.getLoggerResolverMutex.RUnlock()
fake.getOfferMutex.RLock()
defer fake.getOfferMutex.RUnlock()
fake.getPacerMutex.RLock()
defer fake.getPacerMutex.RUnlock()
fake.getPendingTrackMutex.RLock()
@@ -9487,8 +9277,6 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
defer fake.removePublishedTrackMutex.RUnlock()
fake.removeTrackLocalMutex.RLock()
defer fake.removeTrackLocalMutex.RUnlock()
fake.sendConnectResponseMutex.RLock()
defer fake.sendConnectResponseMutex.RUnlock()
fake.sendConnectionQualityUpdateMutex.RLock()
defer fake.sendConnectionQualityUpdateMutex.RUnlock()
fake.sendDataMessageMutex.RLock()
@@ -9535,8 +9323,6 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
defer fake.setSubscriberChannelCapacityMutex.RUnlock()
fake.setTrackMutedMutex.RLock()
defer fake.setTrackMutedMutex.RUnlock()
fake.signalPendingMessagesMutex.RLock()
defer fake.signalPendingMessagesMutex.RUnlock()
fake.stateMutex.RLock()
defer fake.stateMutex.RUnlock()
fake.stopAndGetSubscribedTracksForwarderStateMutex.RLock()
pkg/rtc/utils.go
-22
View File
@@ -171,25 +171,3 @@ func MaybeTruncateIP(addr string) string {
return addr[:len(addr)-3] + "..."
}
func Signalv2FragmentsAsSignalv2WireMessages(fragments []*livekit.Fragment) []*livekit.Signalv2WireMessage {
msgs := make([]*livekit.Signalv2WireMessage, len(fragments))
for idx, fragment := range fragments {
msgs[idx] = &livekit.Signalv2WireMessage{
Message: &livekit.Signalv2WireMessage_Fragment{
Fragment: fragment,
},
}
}
return msgs
}
func EnvelopeSignalv2ServerMessages(msgs []*livekit.Signalv2ServerMessage) *livekit.Signalv2WireMessage {
return &livekit.Signalv2WireMessage{
Message: &livekit.Signalv2WireMessage_Envelope{
Envelope: &livekit.Envelope{
ServerMessages: msgs,
},
},
}
}
pkg/service/roommanager.go
-236
View File
@@ -34,7 +34,6 @@ import (
"github.com/livekit/protocol/observability"
"github.com/livekit/protocol/observability/roomobs"
"github.com/livekit/protocol/rpc"
"github.com/livekit/protocol/signalling"
"github.com/livekit/protocol/utils"
"github.com/livekit/protocol/utils/guid"
"github.com/livekit/protocol/utils/must"
@@ -566,241 +565,6 @@ func (r *RoomManager) StartSession(
return nil
}
// SIGNALLING-V2-TODO: consolidate common parts beteen this and `StartSession`
func (r *RoomManager) HandleConnect(
ctx context.Context,
grants *auth.ClaimGrants,
createRoom *livekit.CreateRoomRequest,
wireMessage *livekit.Signalv2WireMessage, // contains livekit.ConnectRequest
) (*livekit.Signalv2WireMessage, error) {
sessionStartTime := time.Now()
// find connect request in wire message
connectRequest := signalling.GetConnectRequest(wireMessage)
if connectRequest == nil {
return nil, ErrNoConnectRequest
}
room, err := r.getOrCreateRoom(ctx, createRoom)
if err != nil {
return nil, err
}
defer room.Release()
protoRoom, roomInternal := room.ToProto(), room.Internal()
participantIdentity := livekit.ParticipantIdentity(grants.Identity)
// should not error out, error is logged in iceServersForParticipant even if it fails
// since this is used for TURN server credentials, we don't want to fail the request even if there's no TURN for the session
apiKey, _, _ := r.getFirstKeyPair()
/* SIGNALLING-V2-TODO - v2 should not have ICERestart, but leaving reminder here (similar location as `StartSession` code check on this path */
sid := livekit.ParticipantID(guid.New(utils.ParticipantPrefix))
pLogger := rtc.LoggerWithParticipant(
rtc.LoggerWithRoom(logger.GetLogger(), room.Name(), room.ID()),
participantIdentity,
sid,
false,
)
pLogger.Infow(
"starting RTC session",
"room", room.Name(),
"nodeID", r.currentNode.NodeID(),
"numParticipants", room.GetParticipantCount(),
"grants", grants,
"createRoom", logger.Proto(createRoom),
"wireMessage", logger.Proto(wireMessage),
)
clientInfo := connectRequest.ClientInfo
clientConf := r.clientConfManager.GetConfiguration(clientInfo)
rtcConf := *r.rtcConfig
rtcConf.SetBufferFactory(room.GetBufferFactory())
if connectRequest.ConnectionSettings.DisableIceLite {
rtcConf.SettingEngine.SetLite(false)
}
// default allow forceTCP
allowFallback := true
if r.config.RTC.AllowTCPFallback != nil {
allowFallback = *r.config.RTC.AllowTCPFallback
}
// default do not force full reconnect on a publication error
reconnectOnPublicationError := false
if r.config.RTC.ReconnectOnPublicationError != nil {
reconnectOnPublicationError = *r.config.RTC.ReconnectOnPublicationError
}
// default do not force full reconnect on a subscription error
reconnectOnSubscriptionError := false
if r.config.RTC.ReconnectOnSubscriptionError != nil {
reconnectOnSubscriptionError = *r.config.RTC.ReconnectOnSubscriptionError
}
// default do not force full reconnect on a data channel error
reconnectOnDataChannelError := false
if r.config.RTC.ReconnectOnDataChannelError != nil {
reconnectOnDataChannelError = *r.config.RTC.ReconnectOnDataChannelError
}
subscriberAllowPause := r.config.RTC.CongestionControl.AllowPause
if connectRequest.ConnectionSettings.SubscriberAllowPause != nil {
subscriberAllowPause = *connectRequest.ConnectionSettings.SubscriberAllowPause
}
participant, err := rtc.NewParticipant(rtc.ParticipantParams{
Identity: participantIdentity,
Name: livekit.ParticipantName(grants.Name),
SID: sid,
Config: &rtcConf,
AudioConfig: r.config.Audio,
VideoConfig: r.config.Video,
LimitConfig: r.config.Limit,
ProtocolVersion: types.ProtocolVersion(clientInfo.Protocol),
SessionStartTime: sessionStartTime,
Telemetry: r.telemetry,
Trailer: room.Trailer(),
PLIThrottleConfig: r.config.RTC.PLIThrottle,
CongestionControlConfig: r.config.RTC.CongestionControl,
PublishEnabledCodecs: protoRoom.EnabledCodecs,
SubscribeEnabledCodecs: protoRoom.EnabledCodecs,
Grants: grants,
Logger: pLogger,
Reporter: roomobs.NewNoopParticipantSessionReporter(),
ClientConf: clientConf,
ClientInfo: rtc.ClientInfo{ClientInfo: clientInfo},
// SIGNALLING-V@-TODO Region: pi.Region,
AdaptiveStream: connectRequest.ConnectionSettings.AdaptiveStream,
AllowTCPFallback: allowFallback,
TURNSEnabled: r.config.IsTURNSEnabled(),
ParticipantHelper: &roomManagerParticipantHelper{
room: room,
codecRegressionThreshold: r.config.Video.CodecRegressionThreshold,
},
ReconnectOnPublicationError: reconnectOnPublicationError,
ReconnectOnSubscriptionError: reconnectOnSubscriptionError,
ReconnectOnDataChannelError: reconnectOnDataChannelError,
VersionGenerator: r.versionGenerator,
SubscriberAllowPause: subscriberAllowPause,
SubscriptionLimitAudio: r.config.Limit.SubscriptionLimitAudio,
SubscriptionLimitVideo: r.config.Limit.SubscriptionLimitVideo,
PlayoutDelay: roomInternal.GetPlayoutDelay(),
SyncStreams: roomInternal.GetSyncStreams(),
ForwardStats: r.forwardStats,
MetricConfig: r.config.Metric,
DataChannelMaxBufferedAmount: r.config.RTC.DataChannelMaxBufferedAmount,
DatachannelSlowThreshold: r.config.RTC.DatachannelSlowThreshold,
FireOnTrackBySdp: true,
SynchronousLocalCandidatesMode: true,
})
if err != nil {
return nil, err
}
iceConfig := r.setIceConfig(room.Name(), participant)
// join room
opts := rtc.ParticipantOptions{
AutoSubscribe: connectRequest.ConnectionSettings.AutoSubscribe,
}
iceServers := r.iceServersForParticipant(
apiKey,
participant,
iceConfig.PreferenceSubscriber == livekit.ICECandidateType_ICT_TLS,
)
if err := room.Joinv2(participant, &opts, iceServers); err != nil {
pLogger.Errorw("could not join room", err)
_ = participant.Close(true, types.ParticipantCloseReasonJoinFailed, false)
return nil, err
}
// SIGNALLING-V2-TODO: process messages other than ConnecRequest in envelope.
var wireMessageResponse *livekit.Signalv2WireMessage
if pending := participant.SignalPendingMessages(); pending != nil {
var ok bool
if wireMessageResponse, ok = pending.(*livekit.Signalv2WireMessage); !ok {
pLogger.Errorw("could not join room", ErrInvalidMessageType)
_ = participant.Close(true, types.ParticipantCloseReasonJoinFailed, false)
return nil, ErrInvalidMessageType
}
}
if signalling.GetConnectResponse(wireMessageResponse) == nil {
pLogger.Errorw("could not join room", ErrNoConnectResponse)
_ = participant.Close(true, types.ParticipantCloseReasonJoinFailed, false)
return nil, ErrNoConnectResponse
}
var participantServerClosers utils.Closers
participantTopic := rpc.FormatParticipantTopic(room.Name(), participant.Identity())
participantServer := must.Get(rpc.NewTypedParticipantServer(r, r.bus))
participantServerClosers = append(participantServerClosers, utils.CloseFunc(r.participantServers.Replace(participantTopic, participantServer)))
if err := participantServer.RegisterAllParticipantTopics(participantTopic); err != nil {
participantServerClosers.Close()
pLogger.Errorw("could not join register participant topic", err)
_ = participant.Close(true, types.ParticipantCloseReasonMessageBusFailed, false)
return nil, err
}
httpSignalParticipantServer := must.Get(rpc.NewTypedSignalv2ParticipantServer(signalv2ParticipantService{r}, r.bus))
participantServerClosers = append(
participantServerClosers,
utils.CloseFunc(r.httpSignalParticipantServers.Replace(participantTopic, httpSignalParticipantServer)),
)
if err := httpSignalParticipantServer.RegisterAllCommonTopics(participantTopic); err != nil {
participantServerClosers.Close()
pLogger.Errorw("could not join register participant topic for http signalling", err)
_ = participant.Close(true, types.ParticipantCloseReasonMessageBusFailed, false)
return nil, err
}
if err = r.roomStore.StoreParticipant(ctx, room.Name(), participant.ToProto()); err != nil {
pLogger.Errorw("could not store participant", err)
}
persistRoomForParticipantCount := func(proto *livekit.Room) {
if !participant.Hidden() && !room.IsClosed() {
err = r.roomStore.StoreRoom(ctx, proto, room.Internal())
if err != nil {
logger.Errorw("could not store room", err)
}
}
}
// update room store with new numParticipants
persistRoomForParticipantCount(room.ToProto())
clientMeta := &livekit.AnalyticsClientMeta{
Region: r.currentNode.Region(),
Node: string(r.currentNode.NodeID()),
}
r.telemetry.ParticipantJoined(ctx, protoRoom, participant.ToProto(), clientInfo, clientMeta, true)
participant.OnClose(func(p types.LocalParticipant) {
if err := r.roomStore.DeleteParticipant(ctx, room.Name(), p.Identity()); err != nil {
pLogger.Errorw("could not delete participant", err)
}
// update room store with new numParticipants
proto := room.ToProto()
persistRoomForParticipantCount(proto)
r.telemetry.ParticipantLeft(ctx, proto, p.ToProto(), true)
})
participant.OnClaimsChanged(func(participant types.LocalParticipant) {
pLogger.Debugw("refreshing client token after claims change")
if err := r.refreshToken(participant); err != nil {
pLogger.Errorw("could not refresh token", err)
}
})
participant.OnICEConfigChanged(func(participant types.LocalParticipant, iceConfig *livekit.ICEConfig) {
r.iceConfigCache.Put(iceConfigCacheKey{room.Name(), participant.Identity()}, iceConfig)
})
return wireMessageResponse, nil
}
// create the actual room object, to be used on RTC node
func (r *RoomManager) getOrCreateRoom(ctx context.Context, createRoom *livekit.CreateRoomRequest) (*rtc.Room, error) {
roomName := livekit.RoomName(createRoom.Name)
pkg/service/roommanager_service.go
-54
View File
@@ -194,57 +194,3 @@ func (r whipParticipantService) DeleteSession(ctx context.Context, req *rpc.WHIP
}
// --------------------------------
type signalv2ParticipantService struct {
*RoomManager
}
func (s signalv2ParticipantService) RelaySignalv2Participant(ctx context.Context, req *rpc.RelaySignalv2ParticipantRequest) (*rpc.RelaySignalv2ParticipantResponse, error) {
room := s.RoomManager.GetRoom(ctx, livekit.RoomName(req.Room))
if room == nil {
return nil, ErrRoomNotFound
}
lp := room.GetParticipantByID(livekit.ParticipantID(req.ParticipantId))
if lp == nil {
return nil, ErrParticipantNotFound
}
err := lp.HandleSignalMessage(req.WireMessage)
if err != nil {
return nil, err
}
var wireMessage *livekit.Signalv2WireMessage
if pending := lp.SignalPendingMessages(); pending != nil {
var ok bool
if wireMessage, ok = pending.(*livekit.Signalv2WireMessage); !ok {
return nil, ErrInvalidMessageType
}
}
return &rpc.RelaySignalv2ParticipantResponse{
WireMessage: wireMessage,
}, nil
}
func (s signalv2ParticipantService) RelaySignalv2ParticipantDeleteSession(
ctx context.Context,
req *rpc.RelaySignalv2ParticipantDeleteSessionRequest,
) (*emptypb.Empty, error) {
room := s.RoomManager.GetRoom(ctx, livekit.RoomName(req.Room))
if room == nil {
return nil, ErrRoomNotFound
}
lp := room.GetParticipantByID(livekit.ParticipantID(req.ParticipantId))
if lp != nil {
room.RemoveParticipant(
lp.Identity(),
lp.ID(),
types.ParticipantCloseReasonClientRequestLeave,
)
}
return &emptypb.Empty{}, nil
}
pkg/service/rtcv2service.go
-361
View File
@@ -1,361 +0,0 @@
// Copyright 2023 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 service
import (
"encoding/json"
"errors"
"fmt"
"io/ioutil"
"net/http"
"github.com/livekit/livekit-server/pkg/config"
"github.com/livekit/livekit-server/pkg/routing"
"github.com/livekit/livekit-server/pkg/rtc"
"github.com/livekit/livekit-server/pkg/utils"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
"github.com/livekit/protocol/rpc"
"github.com/livekit/protocol/signalling"
"github.com/livekit/psrpc"
"google.golang.org/protobuf/proto"
)
var (
errFragmentsInHTTP = errors.New("should not get fragments via HTTP request")
errUnknownMessageType = errors.New("unknown message type")
)
const (
cRTCv2Path = "/rtc/v2"
cRTCv2ValidatePath = "/rtc/v2/validate"
cRTCv2ParticipantIDPath = "/rtc/v2/{participant_id}"
)
type RTCv2Service struct {
http.Handler
limits config.LimitConfig
roomAllocator RoomAllocator
router routing.MessageRouter
topicFormatter rpc.TopicFormatter
signalv2ParticipantClient rpc.TypedSignalv2ParticipantClient
}
func NewRTCv2Service(
config *config.Config,
roomAllocator RoomAllocator,
router routing.MessageRouter,
topicFormatter rpc.TopicFormatter,
signalv2ParticipantClient rpc.TypedSignalv2ParticipantClient,
) *RTCv2Service {
return &RTCv2Service{
limits: config.Limit,
router: router,
roomAllocator: roomAllocator,
topicFormatter: topicFormatter,
signalv2ParticipantClient: signalv2ParticipantClient,
}
}
func (s *RTCv2Service) SetupRoutes(mux *http.ServeMux) {
mux.HandleFunc("POST "+cRTCv2Path, s.handlePost)
mux.HandleFunc("GET "+cRTCv2ValidatePath, s.validate)
mux.HandleFunc("PATCH "+cRTCv2ParticipantIDPath, s.handleParticipantPatch)
mux.HandleFunc("DELETE "+cRTCv2ParticipantIDPath, s.handleParticipantDelete)
}
func (s *RTCv2Service) validateInternal(
lgr logger.Logger,
r *http.Request,
wireMessage *livekit.Signalv2WireMessage,
) (livekit.RoomName, livekit.ParticipantIdentity, *rpc.RelaySignalv2ConnectRequest, int, error) {
connectRequest := signalling.GetConnectRequest(wireMessage)
if connectRequest == nil {
return "", "", nil, http.StatusBadRequest, ErrNoConnectRequest
}
params := ValidateConnectRequestParams{
metadata: connectRequest.Metadata,
attributes: connectRequest.ParticipantAttributes,
}
res, code, err := ValidateConnectRequest(
lgr,
r,
s.limits,
params,
s.router,
s.roomAllocator,
)
if err != nil {
return "", "", nil, code, err
}
grantsJson, err := json.Marshal(res.grants)
if err != nil {
return "", "", nil, http.StatusInternalServerError, err
}
AugmentClientInfo(connectRequest.ClientInfo, r)
return res.roomName,
livekit.ParticipantIdentity(res.grants.Identity),
&rpc.RelaySignalv2ConnectRequest{
GrantsJson: string(grantsJson),
CreateRoom: res.createRoomRequest,
WireMessage: wireMessage,
},
code,
err
}
func (s *RTCv2Service) handlePost(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("Content-type") != "application/x-protobuf" {
HandleErrorJson(w, r, http.StatusBadRequest, fmt.Errorf("unsupported content-type: %s", r.Header.Get("Content-type")))
return
}
wireMessage, err := getWireMessage(r)
if err != nil {
HandleErrorJson(w, r, http.StatusBadRequest, fmt.Errorf("could not get wire message: %w", err))
return
}
// only connect requests should be coming in here and there should not be fragments
roomName, participantIdentity, rscr, code, err := s.validateInternal(
utils.GetLogger(r.Context()),
r,
wireMessage,
)
if err != nil {
HandleErrorJson(w, r, code, err)
return
}
if err := s.roomAllocator.SelectRoomNode(r.Context(), roomName, ""); err != nil {
HandleErrorJson(w, r, http.StatusInternalServerError, err)
return
}
resp, err := s.router.HandleParticipantConnectRequest(r.Context(), roomName, participantIdentity, rscr)
if err != nil {
HandleErrorJson(w, r, http.StatusInternalServerError, err)
return
}
connectResponse := signalling.GetConnectResponse(resp.WireMessage)
if connectResponse == nil {
HandleErrorJson(w, r, http.StatusInternalServerError, ErrNoConnectResponse)
return
}
marshalled, err := proto.Marshal(resp.WireMessage)
if err != nil {
HandleErrorJson(w, r, http.StatusInternalServerError, err)
return
}
w.Header().Add("Content-type", "application/x-protobuf")
w.Write(marshalled)
logger.Debugw(
"connect response",
"room", roomName,
"roomID", connectResponse.Room.Sid, // SIGNALLING-V2-TODO: roomID may not be resolved
"participant", participantIdentity,
"pID", connectResponse.Participant.Sid,
"wireMessage", logger.Proto(resp.WireMessage),
)
w.WriteHeader(http.StatusOK)
}
func (s *RTCv2Service) validate(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("Content-type") != "application/x-protobuf" {
HandleErrorJson(w, r, http.StatusBadRequest, fmt.Errorf("unsupported content-type: %s", r.Header.Get("Content-type")))
return
}
wireMessage, err := getWireMessage(r)
if err != nil {
HandleErrorJson(w, r, http.StatusBadRequest, fmt.Errorf("could not get wire message: %w", err))
return
}
_, _, _, code, err := s.validateInternal(utils.GetLogger(r.Context()), r, wireMessage)
if err != nil {
HandleErrorJson(w, r, code, err)
return
}
_, _ = w.Write([]byte("success"))
w.WriteHeader(http.StatusOK)
}
func (s *RTCv2Service) handleParticipantPatch(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("Content-type") != "application/x-protobuf" {
HandleErrorJson(w, r, http.StatusBadRequest, fmt.Errorf("unsupported content-type: %s", r.Header.Get("Content-type")))
return
}
roomName, participantIdentity, pID, code, err := getParams(r)
if err != nil {
HandleErrorJson(w, r, code, err)
return
}
wireMessage, err := getWireMessage(r)
if err != nil {
HandleErrorJson(w, r, http.StatusBadRequest, fmt.Errorf("could not get wire message: %w", err))
return
}
logger.Debugw(
"participant request",
"room", roomName,
"participant", participantIdentity,
"pID", pID,
"participantRequest", logger.Proto(wireMessage),
)
res, err := s.signalv2ParticipantClient.RelaySignalv2Participant(
r.Context(),
s.topicFormatter.ParticipantTopic(r.Context(), roomName, participantIdentity),
&rpc.RelaySignalv2ParticipantRequest{
Room: string(roomName),
ParticipantIdentity: string(participantIdentity),
ParticipantId: string(pID),
WireMessage: wireMessage,
},
)
if err != nil {
var pe psrpc.Error
if errors.As(err, &pe) {
switch pe.Code() {
case psrpc.NotFound:
HandleErrorJson(w, r, http.StatusNotFound, errors.New(pe.Error()))
case psrpc.InvalidArgument:
HandleErrorJson(w, r, http.StatusBadRequest, errors.New(pe.Error()))
default:
HandleErrorJson(w, r, http.StatusInternalServerError, errors.New(pe.Error()))
}
} else {
HandleErrorJson(w, r, http.StatusInternalServerError, nil)
}
return
}
logger.Debugw(
"participant response",
"room", roomName,
"participant", participantIdentity,
"pID", pID,
"participantResponse", logger.Proto(res.WireMessage),
)
marshalled, err := proto.Marshal(res.WireMessage)
if err != nil {
HandleErrorJson(w, r, http.StatusInternalServerError, err)
return
}
w.Header().Add("Content-type", "application/x-protobuf")
w.Write(marshalled)
w.WriteHeader(http.StatusOK)
}
func (s *RTCv2Service) handleParticipantDelete(w http.ResponseWriter, r *http.Request) {
claims := GetGrants(r.Context())
if claims == nil || claims.Video == nil {
HandleErrorJson(w, r, http.StatusUnauthorized, rtc.ErrPermissionDenied)
return
}
roomName, participantIdentity, pID, code, err := getParams(r)
if err != nil {
HandleErrorJson(w, r, code, err)
return
}
_, err = s.signalv2ParticipantClient.RelaySignalv2ParticipantDeleteSession(
r.Context(),
s.topicFormatter.ParticipantTopic(r.Context(), roomName, participantIdentity),
&rpc.RelaySignalv2ParticipantDeleteSessionRequest{
Room: string(roomName),
ParticipantIdentity: string(participantIdentity),
ParticipantId: string(pID),
},
)
if err != nil {
HandleErrorJson(w, r, http.StatusBadRequest, err)
return
}
logger.Debugw(
"participant deleted",
"room", roomName,
"participant", participantIdentity,
"pID", pID,
)
w.WriteHeader(http.StatusOK)
}
// ---------------------------------------
func getWireMessage(r *http.Request) (*livekit.Signalv2WireMessage, error) {
body, err := ioutil.ReadAll(r.Body)
if err != nil {
return nil, err
}
wireMessage := &livekit.Signalv2WireMessage{}
err = proto.Unmarshal(body, wireMessage)
if err != nil {
return nil, err
}
return wireMessage, nil
}
func getParams(r *http.Request) (livekit.RoomName, livekit.ParticipantIdentity, livekit.ParticipantID, int, error) {
claims := GetGrants(r.Context())
if claims == nil || claims.Video == nil {
return "", "", "", http.StatusUnauthorized, rtc.ErrPermissionDenied
}
roomName, err := EnsureJoinPermission(r.Context())
if err != nil {
return "", "", "", http.StatusUnauthorized, err
}
if roomName == "" {
return "", "", "", http.StatusUnauthorized, ErrNoRoomName
}
participantIdentity := livekit.ParticipantIdentity(claims.Identity)
if participantIdentity == "" {
return "", "", "", http.StatusUnauthorized, ErrIdentityEmpty
}
pID := livekit.ParticipantID(r.PathValue("participant_id"))
if pID == "" {
return "", "", "", http.StatusBadRequest, ErrParticipantSidEmpty
}
return roomName, participantIdentity, pID, http.StatusOK, nil
}
pkg/service/server.go
+23 -34
View File
@@ -46,23 +46,21 @@ import (
)
type LivekitServer struct {
config *config.Config
ioService *IOInfoService
rtcService *RTCService
rtcv2Service *RTCv2Service
whipService *WHIPService
agentService *AgentService
httpServer *http.Server
promServer *http.Server
router routing.Router
roomManager *RoomManager
signalServer *SignalServer
signalv2Server *Signalv2Server
turnServer *turn.Server
currentNode routing.LocalNode
running atomic.Bool
doneChan chan struct{}
closedChan chan struct{}
config *config.Config
ioService *IOInfoService
rtcService *RTCService
whipService *WHIPService
agentService *AgentService
httpServer *http.Server
promServer *http.Server
router routing.Router
roomManager *RoomManager
signalServer *SignalServer
turnServer *turn.Server
currentNode routing.LocalNode
running atomic.Bool
doneChan chan struct{}
closedChan chan struct{}
}
func NewLivekitServer(conf *config.Config,
@@ -73,28 +71,24 @@ func NewLivekitServer(conf *config.Config,
sipService *SIPService,
ioService *IOInfoService,
rtcService *RTCService,
rtcv2Service *RTCv2Service,
whipService *WHIPService,
agentService *AgentService,
keyProvider auth.KeyProvider,
router routing.Router,
roomManager *RoomManager,
signalServer *SignalServer,
signalv2Server *Signalv2Server,
turnServer *turn.Server,
currentNode routing.LocalNode,
) (s *LivekitServer, err error) {
s = &LivekitServer{
config: conf,
ioService: ioService,
rtcService: rtcService,
rtcv2Service: rtcv2Service,
whipService: whipService,
agentService: agentService,
router: router,
roomManager: roomManager,
signalServer: signalServer,
signalv2Server: signalv2Server,
config: conf,
ioService: ioService,
rtcService: rtcService,
whipService: whipService,
agentService: agentService,
router: router,
roomManager: roomManager,
signalServer: signalServer,
// turn server starts automatically
turnServer: turnServer,
currentNode: currentNode,
@@ -150,7 +144,6 @@ func NewLivekitServer(conf *config.Config,
xtwirp.RegisterServer(mux, ingressServer)
xtwirp.RegisterServer(mux, sipServer)
rtcService.SetupRoutes(mux)
rtcv2Service.SetupRoutes(mux)
whipService.SetupRoutes(mux)
mux.Handle("/agent", agentService)
mux.HandleFunc("/", s.defaultHandler)
@@ -281,10 +274,6 @@ func (s *LivekitServer) Start() error {
return err
}
if err := s.signalv2Server.Start(); err != nil {
return err
}
httpGroup := &errgroup.Group{}
for _, ln := range listeners {
l := ln
pkg/service/servicefakes/fake_connect_handler.go
-200
View File
@@ -1,200 +0,0 @@
// Code generated by counterfeiter. DO NOT EDIT.
package servicefakes
import (
"context"
"sync"
"github.com/livekit/livekit-server/pkg/service"
"github.com/livekit/protocol/auth"
"github.com/livekit/protocol/logger"
"github.com/livekit/protocol/rpc"
)
type FakeConnectHandler struct {
HandleConnectStub func(context.Context, logger.Logger, *auth.ClaimGrants, *rpc.RelaySignalv2ConnectRequest) (*rpc.RelaySignalv2ConnectResponse, error)
handleConnectMutex sync.RWMutex
handleConnectArgsForCall []struct {
arg1 context.Context
arg2 logger.Logger
arg3 *auth.ClaimGrants
arg4 *rpc.RelaySignalv2ConnectRequest
}
handleConnectReturns struct {
result1 *rpc.RelaySignalv2ConnectResponse
result2 error
}
handleConnectReturnsOnCall map[int]struct {
result1 *rpc.RelaySignalv2ConnectResponse
result2 error
}
LoggerStub func(context.Context) logger.Logger
loggerMutex sync.RWMutex
loggerArgsForCall []struct {
arg1 context.Context
}
loggerReturns struct {
result1 logger.Logger
}
loggerReturnsOnCall map[int]struct {
result1 logger.Logger
}
invocations map[string][][]interface{}
invocationsMutex sync.RWMutex
}
func (fake *FakeConnectHandler) HandleConnect(arg1 context.Context, arg2 logger.Logger, arg3 *auth.ClaimGrants, arg4 *rpc.RelaySignalv2ConnectRequest) (*rpc.RelaySignalv2ConnectResponse, error) {
fake.handleConnectMutex.Lock()
ret, specificReturn := fake.handleConnectReturnsOnCall[len(fake.handleConnectArgsForCall)]
fake.handleConnectArgsForCall = append(fake.handleConnectArgsForCall, struct {
arg1 context.Context
arg2 logger.Logger
arg3 *auth.ClaimGrants
arg4 *rpc.RelaySignalv2ConnectRequest
}{arg1, arg2, arg3, arg4})
stub := fake.HandleConnectStub
fakeReturns := fake.handleConnectReturns
fake.recordInvocation("HandleConnect", []interface{}{arg1, arg2, arg3, arg4})
fake.handleConnectMutex.Unlock()
if stub != nil {
return stub(arg1, arg2, arg3, arg4)
}
if specificReturn {
return ret.result1, ret.result2
}
return fakeReturns.result1, fakeReturns.result2
}
func (fake *FakeConnectHandler) HandleConnectCallCount() int {
fake.handleConnectMutex.RLock()
defer fake.handleConnectMutex.RUnlock()
return len(fake.handleConnectArgsForCall)
}
func (fake *FakeConnectHandler) HandleConnectCalls(stub func(context.Context, logger.Logger, *auth.ClaimGrants, *rpc.RelaySignalv2ConnectRequest) (*rpc.RelaySignalv2ConnectResponse, error)) {
fake.handleConnectMutex.Lock()
defer fake.handleConnectMutex.Unlock()
fake.HandleConnectStub = stub
}
func (fake *FakeConnectHandler) HandleConnectArgsForCall(i int) (context.Context, logger.Logger, *auth.ClaimGrants, *rpc.RelaySignalv2ConnectRequest) {
fake.handleConnectMutex.RLock()
defer fake.handleConnectMutex.RUnlock()
argsForCall := fake.handleConnectArgsForCall[i]
return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4
}
func (fake *FakeConnectHandler) HandleConnectReturns(result1 *rpc.RelaySignalv2ConnectResponse, result2 error) {
fake.handleConnectMutex.Lock()
defer fake.handleConnectMutex.Unlock()
fake.HandleConnectStub = nil
fake.handleConnectReturns = struct {
result1 *rpc.RelaySignalv2ConnectResponse
result2 error
}{result1, result2}
}
func (fake *FakeConnectHandler) HandleConnectReturnsOnCall(i int, result1 *rpc.RelaySignalv2ConnectResponse, result2 error) {
fake.handleConnectMutex.Lock()
defer fake.handleConnectMutex.Unlock()
fake.HandleConnectStub = nil
if fake.handleConnectReturnsOnCall == nil {
fake.handleConnectReturnsOnCall = make(map[int]struct {
result1 *rpc.RelaySignalv2ConnectResponse
result2 error
})
}
fake.handleConnectReturnsOnCall[i] = struct {
result1 *rpc.RelaySignalv2ConnectResponse
result2 error
}{result1, result2}
}
func (fake *FakeConnectHandler) Logger(arg1 context.Context) logger.Logger {
fake.loggerMutex.Lock()
ret, specificReturn := fake.loggerReturnsOnCall[len(fake.loggerArgsForCall)]
fake.loggerArgsForCall = append(fake.loggerArgsForCall, struct {
arg1 context.Context
}{arg1})
stub := fake.LoggerStub
fakeReturns := fake.loggerReturns
fake.recordInvocation("Logger", []interface{}{arg1})
fake.loggerMutex.Unlock()
if stub != nil {
return stub(arg1)
}
if specificReturn {
return ret.result1
}
return fakeReturns.result1
}
func (fake *FakeConnectHandler) LoggerCallCount() int {
fake.loggerMutex.RLock()
defer fake.loggerMutex.RUnlock()
return len(fake.loggerArgsForCall)
}
func (fake *FakeConnectHandler) LoggerCalls(stub func(context.Context) logger.Logger) {
fake.loggerMutex.Lock()
defer fake.loggerMutex.Unlock()
fake.LoggerStub = stub
}
func (fake *FakeConnectHandler) LoggerArgsForCall(i int) context.Context {
fake.loggerMutex.RLock()
defer fake.loggerMutex.RUnlock()
argsForCall := fake.loggerArgsForCall[i]
return argsForCall.arg1
}
func (fake *FakeConnectHandler) LoggerReturns(result1 logger.Logger) {
fake.loggerMutex.Lock()
defer fake.loggerMutex.Unlock()
fake.LoggerStub = nil
fake.loggerReturns = struct {
result1 logger.Logger
}{result1}
}
func (fake *FakeConnectHandler) LoggerReturnsOnCall(i int, result1 logger.Logger) {
fake.loggerMutex.Lock()
defer fake.loggerMutex.Unlock()
fake.LoggerStub = nil
if fake.loggerReturnsOnCall == nil {
fake.loggerReturnsOnCall = make(map[int]struct {
result1 logger.Logger
})
}
fake.loggerReturnsOnCall[i] = struct {
result1 logger.Logger
}{result1}
}
func (fake *FakeConnectHandler) Invocations() map[string][][]interface{} {
fake.invocationsMutex.RLock()
defer fake.invocationsMutex.RUnlock()
fake.handleConnectMutex.RLock()
defer fake.handleConnectMutex.RUnlock()
fake.loggerMutex.RLock()
defer fake.loggerMutex.RUnlock()
copiedInvocations := map[string][][]interface{}{}
for key, value := range fake.invocations {
copiedInvocations[key] = value
}
return copiedInvocations
}
func (fake *FakeConnectHandler) recordInvocation(key string, args []interface{}) {
fake.invocationsMutex.Lock()
defer fake.invocationsMutex.Unlock()
if fake.invocations == nil {
fake.invocations = map[string][][]interface{}{}
}
if fake.invocations[key] == nil {
fake.invocations[key] = [][]interface{}{}
}
fake.invocations[key] = append(fake.invocations[key], args)
}
var _ service.ConnectHandler = new(FakeConnectHandler)
pkg/service/signalv2.go
-162
View File
@@ -1,162 +0,0 @@
// Copyright 2023 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 service
import (
"context"
"encoding/json"
"github.com/livekit/livekit-server/pkg/routing"
"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
"github.com/livekit/livekit-server/pkg/utils"
"github.com/livekit/protocol/auth"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
"github.com/livekit/protocol/rpc"
"github.com/livekit/psrpc"
"github.com/livekit/psrpc/pkg/middleware"
)
//counterfeiter:generate . ConnectHandler
type ConnectHandler interface {
Logger(ctx context.Context) logger.Logger
HandleConnect(
ctx context.Context,
lgr logger.Logger,
grants *auth.ClaimGrants,
rscr *rpc.RelaySignalv2ConnectRequest,
) (*rpc.RelaySignalv2ConnectResponse, error)
}
type Signalv2Server struct {
server rpc.TypedSignalv2Server
nodeID livekit.NodeID
}
func NewSignalv2Server(
nodeID livekit.NodeID,
region string,
bus psrpc.MessageBus,
connectHandler ConnectHandler,
) (*Signalv2Server, error) {
s, err := rpc.NewTypedSignalv2Server(
nodeID,
&signalv2Service{region, connectHandler},
bus,
middleware.WithServerMetrics(rpc.PSRPCMetricsObserver{}),
)
if err != nil {
return nil, err
}
return &Signalv2Server{s, nodeID}, nil
}
func (s *Signalv2Server) Start() error {
logger.Debugw("starting relay signal v2 server", "topic", s.nodeID)
return s.server.RegisterAllNodeTopics(s.nodeID)
}
func (r *Signalv2Server) Stop() {
r.server.Kill()
}
// -------------------------------------------------
func NewDefaultSignalv2Server(
currentNode routing.LocalNode,
bus psrpc.MessageBus,
router routing.Router,
roomManager *RoomManager,
) (*Signalv2Server, error) {
return NewSignalv2Server(
currentNode.NodeID(),
currentNode.Region(),
bus,
&defaultSignalv2Handler{currentNode, router, roomManager},
)
}
// -------------------------------------------------
type defaultSignalv2Handler struct {
currentNode routing.LocalNode
router routing.Router
roomManager *RoomManager
}
func (s *defaultSignalv2Handler) Logger(ctx context.Context) logger.Logger {
return utils.GetLogger(ctx)
}
func (s *defaultSignalv2Handler) HandleConnect(
ctx context.Context,
lgr logger.Logger,
grants *auth.ClaimGrants,
rscr *rpc.RelaySignalv2ConnectRequest,
) (*rpc.RelaySignalv2ConnectResponse, error) {
prometheus.IncrementParticipantRtcInit(1)
rtcNode, err := s.router.GetNodeForRoom(ctx, livekit.RoomName(rscr.CreateRoom.Name))
if err != nil {
return nil, err
}
if livekit.NodeID(rtcNode.Id) != s.currentNode.NodeID() {
err = routing.ErrIncorrectRTCNode
lgr.Errorw(
"called participant on incorrect node", err,
"rtcNode", rtcNode,
)
return nil, err
}
wireMessageResponse, err := s.roomManager.HandleConnect(ctx, grants, rscr.CreateRoom, rscr.WireMessage)
if err != nil {
return nil, err
}
return &rpc.RelaySignalv2ConnectResponse{
WireMessage: wireMessageResponse,
}, nil
}
// ------------------------------------------
type signalv2Service struct {
region string
connectHandler ConnectHandler
}
func (r *signalv2Service) RelaySignalv2Connect(ctx context.Context, rscr *rpc.RelaySignalv2ConnectRequest) (*rpc.RelaySignalv2ConnectResponse, error) {
grants := &auth.ClaimGrants{}
if err := json.Unmarshal([]byte(rscr.GrantsJson), grants); err != nil {
return nil, err
}
lgr := r.connectHandler.Logger(ctx).WithValues(
"room", grants.Video.Room,
"participant", grants.Identity,
// SIGNALLING-V2-TODO - maybe add a connection ID to rscr for tracking/debugging only
)
resp, err := r.connectHandler.HandleConnect(ctx, lgr, grants, rscr)
if err != nil {
lgr.Errorw("could not handle new participant", err)
}
return resp, err
}
// ------------------------------------------
pkg/service/wire.go
-3
View File
@@ -77,7 +77,6 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
NewRoomAllocator,
NewRoomService,
NewRTCService,
NewRTCv2Service,
NewWHIPService,
NewAgentService,
NewAgentDispatchService,
@@ -85,7 +84,6 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
getAgentStore,
getSignalRelayConfig,
NewDefaultSignalServer,
NewDefaultSignalv2Server,
routing.NewSignalClient,
getRoomConfig,
routing.NewRoomManagerClient,
@@ -95,7 +93,6 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
rpc.NewTopicFormatter,
rpc.NewTypedRoomClient,
rpc.NewTypedParticipantClient,
rpc.NewTypedSignalv2ParticipantClient,
rpc.NewTypedWHIPParticipantClient,
rpc.NewTypedAgentDispatchInternalClient,
NewLocalRoomManager,
pkg/service/wire_gen.go
+8 -17
View File
@@ -89,23 +89,23 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
}
rtcEgressLauncher := NewEgressLauncher(egressClient, ioInfoService, objectStore)
topicFormatter := rpc.NewTopicFormatter()
roomClient, err := rpc.NewTypedRoomClient(clientParams)
v, err := rpc.NewTypedRoomClient(clientParams)
if err != nil {
return nil, err
}
participantClient, err := rpc.NewTypedParticipantClient(clientParams)
v2, err := rpc.NewTypedParticipantClient(clientParams)
if err != nil {
return nil, err
}
roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, rtcEgressLauncher, topicFormatter, roomClient, participantClient)
roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, rtcEgressLauncher, topicFormatter, v, v2)
if err != nil {
return nil, err
}
agentDispatchInternalClient, err := rpc.NewTypedAgentDispatchInternalClient(clientParams)
v3, err := rpc.NewTypedAgentDispatchInternalClient(clientParams)
if err != nil {
return nil, err
}
agentDispatchService := NewAgentDispatchService(agentDispatchInternalClient, topicFormatter, roomAllocator, router)
agentDispatchService := NewAgentDispatchService(v3, topicFormatter, roomAllocator, router)
egressService := NewEgressService(egressClient, rtcEgressLauncher, ioInfoService, roomService)
ingressConfig := getIngressConfig(conf)
ingressClient, err := rpc.NewIngressClient(clientParams)
@@ -120,16 +120,11 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
}
sipService := NewSIPService(sipConfig, nodeID, messageBus, sipClient, sipStore, roomService, telemetryService)
rtcService := NewRTCService(conf, roomAllocator, router, telemetryService)
signalv2ParticipantClient, err := rpc.NewTypedSignalv2ParticipantClient(clientParams)
v4, err := rpc.NewTypedWHIPParticipantClient(clientParams)
if err != nil {
return nil, err
}
rtCv2Service := NewRTCv2Service(conf, roomAllocator, router, topicFormatter, signalv2ParticipantClient)
whipParticipantClient, err := rpc.NewTypedWHIPParticipantClient(clientParams)
if err != nil {
return nil, err
}
serviceWHIPService, err := NewWHIPService(conf, router, roomAllocator, clientParams, topicFormatter, whipParticipantClient)
serviceWHIPService, err := NewWHIPService(conf, router, roomAllocator, clientParams, topicFormatter, v4)
if err != nil {
return nil, err
}
@@ -153,16 +148,12 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
if err != nil {
return nil, err
}
signalv2Server, err := NewDefaultSignalv2Server(currentNode, messageBus, router, roomManager)
if err != nil {
return nil, err
}
authHandler := getTURNAuthHandlerFunc(turnAuthHandler)
server, err := newInProcessTurnServer(conf, authHandler)
if err != nil {
return nil, err
}
livekitServer, err := NewLivekitServer(conf, roomService, agentDispatchService, egressService, ingressService, sipService, ioInfoService, rtcService, rtCv2Service, serviceWHIPService, agentService, keyProvider, router, roomManager, signalServer, signalv2Server, server, currentNode)
livekitServer, err := NewLivekitServer(conf, roomService, agentDispatchService, egressService, ingressService, sipService, ioInfoService, rtcService, serviceWHIPService, agentService, keyProvider, router, roomManager, signalServer, server, currentNode)
if err != nil {
return nil, err
}