diff --git a/go.mod b/go.mod index d80200627..f94bfbc4f 100644 --- a/go.mod +++ b/go.mod @@ -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 ) diff --git a/go.sum b/go.sum index ebdd2b36c..e920532ee 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/pkg/routing/interfaces.go b/pkg/routing/interfaces.go index 9b7a9dbb3..7550960c1 100644 --- a/pkg/routing/interfaces.go +++ b/pkg/routing/interfaces.go @@ -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( diff --git a/pkg/routing/localrouter.go b/pkg/routing/localrouter.go index 6c09f4236..f648f744f 100644 --- a/pkg/routing/localrouter.go +++ b/pkg/routing/localrouter.go @@ -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 diff --git a/pkg/routing/redisrouter.go b/pkg/routing/redisrouter.go index d7e4039bf..44859b4db 100644 --- a/pkg/routing/redisrouter.go +++ b/pkg/routing/redisrouter.go @@ -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 diff --git a/pkg/routing/routingfakes/fake_router.go b/pkg/routing/routingfakes/fake_router.go index 3b8356a6e..ca8377511 100644 --- a/pkg/routing/routingfakes/fake_router.go +++ b/pkg/routing/routingfakes/fake_router.go @@ -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() diff --git a/pkg/routing/routingfakes/fake_signal_client.go b/pkg/routing/routingfakes/fake_signal_client.go index 586df9a52..0562b7c44 100644 --- a/pkg/routing/routingfakes/fake_signal_client.go +++ b/pkg/routing/routingfakes/fake_signal_client.go @@ -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{}{} diff --git a/pkg/routing/signal.go b/pkg/routing/signal.go index 1f52db964..3f296b931 100644 --- a/pkg/routing/signal.go +++ b/pkg/routing/signal.go @@ -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{} diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index f56855fb1..bf5d9bc0a 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -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 diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go index 2e08d40c5..53574ded4 100644 --- a/pkg/rtc/participant_signal.go +++ b/pkg/rtc/participant_signal.go @@ -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() -} diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 2989f4b87..a9de18a03 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -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 diff --git a/pkg/rtc/signalling/datachannel_messagesink.go b/pkg/rtc/signalling/datachannel_messagesink.go deleted file mode 100644 index 7848af611..000000000 --- a/pkg/rtc/signalling/datachannel_messagesink.go +++ /dev/null @@ -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 "" -} diff --git a/pkg/rtc/signalling/interfaces.go b/pkg/rtc/signalling/interfaces.go index 4b31b92ec..0dd94a7f7 100644 --- a/pkg/rtc/signalling/interfaces.go +++ b/pkg/rtc/signalling/interfaces.go @@ -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 } diff --git a/pkg/rtc/signalling/signalhandler.go b/pkg/rtc/signalling/signalhandler.go index 4e4e08c65..bdc42f85d 100644 --- a/pkg/rtc/signalling/signalhandler.go +++ b/pkg/rtc/signalling/signalhandler.go @@ -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) -} diff --git a/pkg/rtc/signalling/signalhandlerunimplemented.go b/pkg/rtc/signalling/signalhandlerunimplemented.go index 91fb85502..9df37bd9a 100644 --- a/pkg/rtc/signalling/signalhandlerunimplemented.go +++ b/pkg/rtc/signalling/signalhandlerunimplemented.go @@ -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() {} diff --git a/pkg/rtc/signalling/signalhandlerv2.go b/pkg/rtc/signalling/signalhandlerv2.go deleted file mode 100644 index b6e8a4421..000000000 --- a/pkg/rtc/signalling/signalhandlerv2.go +++ /dev/null @@ -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() -} diff --git a/pkg/rtc/signalling/signallerasync.go b/pkg/rtc/signalling/signallerasync.go index 080cf3194..2a0b018b3 100644 --- a/pkg/rtc/signalling/signallerasync.go +++ b/pkg/rtc/signalling/signallerasync.go @@ -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( diff --git a/pkg/rtc/signalling/signallerv2async.go b/pkg/rtc/signalling/signallerv2async.go deleted file mode 100644 index 578a1853d..000000000 --- a/pkg/rtc/signalling/signallerv2async.go +++ /dev/null @@ -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 -} diff --git a/pkg/rtc/signalling/signallerv2hybrid.go b/pkg/rtc/signalling/signallerv2hybrid.go deleted file mode 100644 index 01c4fddd0..000000000 --- a/pkg/rtc/signalling/signallerv2hybrid.go +++ /dev/null @@ -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), - } -} diff --git a/pkg/rtc/signalling/signallingunimplemented.go b/pkg/rtc/signalling/signallingunimplemented.go index eefd011e5..d069d5e2d 100644 --- a/pkg/rtc/signalling/signallingunimplemented.go +++ b/pkg/rtc/signalling/signallingunimplemented.go @@ -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 -} diff --git a/pkg/rtc/signalling/signallingv2.go b/pkg/rtc/signalling/signallingv2.go deleted file mode 100644 index 393abed9b..000000000 --- a/pkg/rtc/signalling/signallingv2.go +++ /dev/null @@ -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}, - }, - }, - } -} diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index b56f6c17e..bf40e948d 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -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 { diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index 613a3eda7..c10fd0840 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -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 { diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 61f3a4e69..2e299bcf0 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -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)) diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index ab1ba0dfb..108a0df49 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -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() diff --git a/pkg/rtc/utils.go b/pkg/rtc/utils.go index ed683840d..9d77153f2 100644 --- a/pkg/rtc/utils.go +++ b/pkg/rtc/utils.go @@ -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, - }, - }, - } -} diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 4e9231aca..a3a680167 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -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) diff --git a/pkg/service/roommanager_service.go b/pkg/service/roommanager_service.go index 41b2f548d..f10812f32 100644 --- a/pkg/service/roommanager_service.go +++ b/pkg/service/roommanager_service.go @@ -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 -} diff --git a/pkg/service/rtcv2service.go b/pkg/service/rtcv2service.go deleted file mode 100644 index 92bd61e38..000000000 --- a/pkg/service/rtcv2service.go +++ /dev/null @@ -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 -} diff --git a/pkg/service/server.go b/pkg/service/server.go index 1d3ecab2e..ffd7b7788 100644 --- a/pkg/service/server.go +++ b/pkg/service/server.go @@ -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 diff --git a/pkg/service/servicefakes/fake_connect_handler.go b/pkg/service/servicefakes/fake_connect_handler.go deleted file mode 100644 index 6ff8030e3..000000000 --- a/pkg/service/servicefakes/fake_connect_handler.go +++ /dev/null @@ -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) diff --git a/pkg/service/signalv2.go b/pkg/service/signalv2.go deleted file mode 100644 index 26416e88f..000000000 --- a/pkg/service/signalv2.go +++ /dev/null @@ -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 -} - -// ------------------------------------------ diff --git a/pkg/service/wire.go b/pkg/service/wire.go index 1757e031e..c079ef126 100644 --- a/pkg/service/wire.go +++ b/pkg/service/wire.go @@ -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, diff --git a/pkg/service/wire_gen.go b/pkg/service/wire_gen.go index 2ac29ab5e..0c62abbbf 100644 --- a/pkg/service/wire_gen.go +++ b/pkg/service/wire_gen.go @@ -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 }