mirror of
https://github.com/livekit/livekit.git
synced 2026-10-22 21:53:22 +00:00
Signalling V2 protocol implementation start (#3794)
* WIP * name * refactor validate * WIP * WIP * signal cache initial impl * HandleConnect in room manager * generate subscriber offer * handle ConnectRequest as stand alone * segmentation, reassembly * clean up * rearrange * lock scope * support metadata in connect request * prom * add SifTrailer to ConnectResponse * prom for get offer error counter * RtcInit counter * Jie feedback * signal client * consolidate v1 and v2 into SignalClient * clean up * comment * deps * mage generate * fix tests * pass around roomName and participantIdentity * mage generate
This commit is contained in:
Raja Subramanian authored and GitHub committed
1 parent
ba702a5323
commit
b9a44c3fbf
32 files changed
+2300
-395
No files matched your search
@@ -153,8 +153,21 @@ type StartParticipantSignalResults struct {
|
||||
type MessageRouter interface {
|
||||
// CreateRoom starts an rtc room
|
||||
CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (res *livekit.Room, err error)
|
||||
|
||||
// StartParticipantSignal participant signal connection is ready to start
|
||||
StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (res StartParticipantSignalResults, err error)
|
||||
StartParticipantSignal(
|
||||
ctx context.Context,
|
||||
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(
|
||||
@@ -229,10 +242,9 @@ func (pi *ParticipantInit) ToStartSession(roomName livekit.RoomName, connectionI
|
||||
}
|
||||
|
||||
ss := &livekit.StartSession{
|
||||
RoomName: string(roomName),
|
||||
Identity: string(pi.Identity),
|
||||
Name: string(pi.Name),
|
||||
// connection id is to allow the RTC node to identify where to route the message back to
|
||||
RoomName: string(roomName),
|
||||
Identity: string(pi.Identity),
|
||||
Name: string(pi.Name),
|
||||
ConnectionId: string(connectionID),
|
||||
Reconnect: pi.Reconnect,
|
||||
ReconnectReason: pi.ReconnectReason,
|
||||
|
||||
@@ -24,6 +24,7 @@ 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)
|
||||
@@ -110,7 +111,8 @@ func (r *LocalRouter) StartParticipantSignal(ctx context.Context, roomName livek
|
||||
func (r *LocalRouter) StartParticipantSignalWithNodeID(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit, nodeID livekit.NodeID) (res StartParticipantSignalResults, err error) {
|
||||
connectionID, reqSink, resSource, err := r.signalClient.StartParticipantSignal(ctx, roomName, pi, nodeID)
|
||||
if err != nil {
|
||||
logger.Errorw("could not handle new participant", err,
|
||||
logger.Errorw(
|
||||
"could not handle new participant", err,
|
||||
"room", roomName,
|
||||
"participant", pi.Identity,
|
||||
"connID", connectionID,
|
||||
@@ -126,6 +128,34 @@ 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
|
||||
|
||||
@@ -169,6 +169,27 @@ 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
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
|
||||
"github.com/livekit/livekit-server/pkg/routing"
|
||||
"github.com/livekit/protocol/livekit"
|
||||
"github.com/livekit/protocol/rpc"
|
||||
)
|
||||
|
||||
type FakeRouter struct {
|
||||
@@ -64,6 +65,22 @@ 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 {
|
||||
@@ -421,6 +438,73 @@ 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)]
|
||||
@@ -855,6 +939,8 @@ 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()
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
|
||||
"github.com/livekit/livekit-server/pkg/routing"
|
||||
"github.com/livekit/protocol/livekit"
|
||||
"github.com/livekit/protocol/rpc"
|
||||
)
|
||||
|
||||
type FakeSignalClient struct {
|
||||
@@ -20,6 +21,23 @@ 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 {
|
||||
@@ -97,6 +115,74 @@ 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)]
|
||||
@@ -175,6 +261,8 @@ 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{}{}
|
||||
|
||||
+57
-8
@@ -41,17 +41,26 @@ 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
|
||||
active atomic.Int32
|
||||
nodeID livekit.NodeID
|
||||
config config.SignalRelayConfig
|
||||
client rpc.TypedSignalClient
|
||||
clientv2 rpc.TypedSignalv2Client
|
||||
active atomic.Int32
|
||||
}
|
||||
|
||||
func NewSignalClient(nodeID livekit.NodeID, bus psrpc.MessageBus, config config.SignalRelayConfig) (SignalClient, error) {
|
||||
c, err := rpc.NewTypedSignalClient(
|
||||
client, err := rpc.NewTypedSignalClient(
|
||||
nodeID,
|
||||
bus,
|
||||
middleware.WithClientMetrics(rpc.PSRPCMetricsObserver{}),
|
||||
@@ -61,10 +70,20 @@ 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: c,
|
||||
nodeID: nodeID,
|
||||
config: config,
|
||||
client: client,
|
||||
clientv2: clientv2,
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -142,6 +161,28 @@ 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{}
|
||||
|
||||
func (e signalRequestMessageWriter) Write(seq uint64, close bool, msgs []proto.Message) *rpc.RelaySignalRequest {
|
||||
@@ -156,6 +197,8 @@ func (e signalRequestMessageWriter) Write(seq uint64, close bool, msgs []proto.M
|
||||
return r
|
||||
}
|
||||
|
||||
// -------------------------------
|
||||
|
||||
type signalResponseMessageReader struct{}
|
||||
|
||||
func (e signalResponseMessageReader) Read(rm *rpc.RelaySignalResponse) ([]proto.Message, error) {
|
||||
@@ -166,6 +209,8 @@ func (e signalResponseMessageReader) Read(rm *rpc.RelaySignalResponse) ([]proto.
|
||||
return msgs, nil
|
||||
}
|
||||
|
||||
// -----------------------------------------
|
||||
|
||||
type RelaySignalMessage interface {
|
||||
proto.Message
|
||||
GetSeq() uint64
|
||||
@@ -212,6 +257,8 @@ func CopySignalStreamToMessageChannel[SendType, RecvType RelaySignalMessage](
|
||||
return stream.Err()
|
||||
}
|
||||
|
||||
// ----------------------------------------
|
||||
|
||||
type signalMessageReader[SendType, RecvType RelaySignalMessage] struct {
|
||||
seq uint64
|
||||
reader SignalMessageReader[RecvType]
|
||||
@@ -239,6 +286,8 @@ func (r *signalMessageReader[SendType, RecvType]) Read(msg RecvType) ([]proto.Me
|
||||
return res, nil
|
||||
}
|
||||
|
||||
// ----------------------------------------
|
||||
|
||||
type SignalSinkParams[SendType, RecvType RelaySignalMessage] struct {
|
||||
Stream psrpc.Stream[SendType, RecvType]
|
||||
Logger logger.Logger
|
||||
|
||||
Reference in new issue
Block a user