dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-28 05:04:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

fixed multi-node routing, correctly set rtc and signal nodes

Browse Source
This commit is contained in:
David Zhao
2021-01-27 01:16:08 -08:00
parent 61db4f5b66
commit 25d476b8ac
10 changed files with 136 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/server/main.go
+1 -1
View File
@@ -185,7 +185,7 @@ func createRouterAndStore(config *config.Config, node routing.LocalNode) (router
return
}
router = routing.NewRedisRouter(node, rc, false)
router = routing.NewRedisRouter(node, rc)
store = service.NewRedisRoomStore(rc)
} else {
// local routing and store
pkg/routing/errors.go
+1
View File
@@ -6,6 +6,7 @@ var (
ErrNotFound = errors.New("could not find object")
ErrHandlerNotDefined = errors.New("handler not defined")
ErrNoAvailableNodes = errors.New("could not find any available nodes")
ErrIncorrectRTCNode = errors.New("current node isn't the RTC node for the room")
errInvalidRouterMessage = errors.New("invalid router message")
ErrChannelClosed = errors.New("channel closed")
)
pkg/routing/interfaces.go
+4 -4
View File
@@ -36,14 +36,14 @@ type Router interface {
GetNode(nodeId string) (*livekit.Node, error)
ListNodes() ([]*livekit.Node, error)
SetParticipantRTCNode(participantId, nodeId string) error
// functions for websocket handler
GetRequestSink(participantId string) (MessageSink, error)
GetResponseSource(participantId string) (MessageSource, error)
StartParticipant(roomName, participantId, participantName string) error
// participant signal connection is ready to start
StartParticipantSignal(roomName, participantId, participantName string) error
OnNewParticipant(callback ParticipantCallback)
// when a new participant's RTC connection is ready to start
OnNewParticipantRTC(callback ParticipantCallback)
Start() error
Stop()
}
pkg/routing/localrouter.go
+5 -13
View File
@@ -10,7 +10,7 @@ import (
// a router of messages on the same node, basic implementation for local testing
type LocalRouter struct {
currentNode LocalNode
lock sync.RWMutex
lock sync.Mutex
// channels for each participant
requestChannels map[string]*MessageChannel
responseChannels map[string]*MessageChannel
@@ -20,7 +20,6 @@ type LocalRouter struct {
func NewLocalRouter(currentNode LocalNode) *LocalRouter {
return &LocalRouter{
currentNode: currentNode,
lock: sync.RWMutex{},
requestChannels: make(map[string]*MessageChannel),
responseChannels: make(map[string]*MessageChannel),
}
@@ -60,7 +59,7 @@ func (r *LocalRouter) ListNodes() ([]*livekit.Node, error) {
}, nil
}
func (r *LocalRouter) StartParticipant(roomName, participantId, participantName string) error {
func (r *LocalRouter) StartParticipantSignal(roomName, participantId, participantName string) error {
// treat it as a new participant connecting
if r.onNewParticipant == nil {
return ErrHandlerNotDefined
@@ -77,11 +76,6 @@ func (r *LocalRouter) StartParticipant(roomName, participantId, participantName
return nil
}
func (r *LocalRouter) SetParticipantRTCNode(participantId, nodeId string) error {
// nothing to be done
return nil
}
// for a local router, sink and source are pointing to the same spot
func (r *LocalRouter) GetRequestSink(participantId string) (MessageSink, error) {
return r.getOrCreateMessageChannel(r.requestChannels, participantId), nil
@@ -91,7 +85,7 @@ func (r *LocalRouter) GetResponseSource(participantId string) (MessageSource, er
return r.getOrCreateMessageChannel(r.responseChannels, participantId), nil
}
func (r *LocalRouter) OnNewParticipant(callback ParticipantCallback) {
func (r *LocalRouter) OnNewParticipantRTC(callback ParticipantCallback) {
r.onNewParticipant = callback
}
@@ -113,9 +107,9 @@ func (r *LocalRouter) statsWorker() {
}
func (r *LocalRouter) getOrCreateMessageChannel(target map[string]*MessageChannel, participantId string) *MessageChannel {
r.lock.RLock()
r.lock.Lock()
defer r.lock.Unlock()
mc := target[participantId]
r.lock.RUnlock()
if mc != nil {
return mc
@@ -127,9 +121,7 @@ func (r *LocalRouter) getOrCreateMessageChannel(target map[string]*MessageChanne
delete(target, participantId)
r.lock.Unlock()
})
r.lock.Lock()
target[participantId] = mc
r.lock.Unlock()
return mc
}
pkg/routing/redisrouter.go
+48 -40
View File
@@ -25,20 +25,18 @@ const (
// Because
type RedisRouter struct {
LocalRouter
useLocal bool
rc *redis.Client
cr *utils.CachedRedis
ctx context.Context
once sync.Once
rc *redis.Client
cr *utils.CachedRedis
ctx context.Context
once sync.Once
redisSinks map[string]*RedisSink
cancel func()
}
func NewRedisRouter(currentNode LocalNode, rc *redis.Client, useLocal bool) *RedisRouter {
func NewRedisRouter(currentNode LocalNode, rc *redis.Client) *RedisRouter {
rr := &RedisRouter{
LocalRouter: *NewLocalRouter(currentNode),
useLocal: useLocal,
rc: rc,
once: sync.Once{},
redisSinks: make(map[string]*RedisSink),
@@ -112,15 +110,6 @@ func (r *RedisRouter) ListNodes() ([]*livekit.Node, error) {
return nodes, nil
}
func (r *RedisRouter) SetParticipantRTCNode(participantId, nodeId string) error {
r.cr.Expire(participantRTCKey(participantId))
err := r.rc.Set(r.ctx, participantRTCKey(participantId), nodeId, participantMappingTTL).Err()
if err != nil {
err = errors.Wrap(err, "could not set rtc node")
}
return err
}
// for a local router, sink and source are pointing to the same spot
func (r *RedisRouter) GetRequestSink(participantId string) (MessageSink, error) {
// request should go to RTC node
@@ -129,47 +118,57 @@ func (r *RedisRouter) GetRequestSink(participantId string) (MessageSink, error)
return nil, err
}
if rtcNode == r.currentNode.Id && r.useLocal {
return r.LocalRouter.GetRequestSink(participantId)
}
sink := r.getOrCreateRedisSink(rtcNode, participantId)
return sink, nil
}
func (r *RedisRouter) GetResponseSource(participantId string) (MessageSource, error) {
// request should go to RTC node
rtcNode, err := r.getParticipantRTCNode(participantId)
if err != nil {
return nil, err
}
if rtcNode == r.currentNode.Id && r.useLocal {
return r.LocalRouter.GetResponseSource(participantId)
}
// a message channel that we'll send data into
source := r.getOrCreateMessageChannel(r.responseChannels, participantId)
return source, nil
}
// StartParticipant always called on the signal node
func (r *RedisRouter) StartParticipant(roomName, participantId, participantName string) error {
// signal connection sets up paths to the RTC node, and starts to route messages to that message queue
func (r *RedisRouter) StartParticipantSignal(roomName, participantId, participantName string) error {
// find the node where the room is hosted at
rtcNode, err := r.GetNodeForRoom(roomName)
if err != nil {
return err
}
if r.useLocal && rtcNode == r.currentNode.Id {
return r.LocalRouter.StartParticipant(roomName, participantId, participantId)
// map signal & rtc nodes
if err = r.setParticipantSignalNode(participantId, r.currentNode.Id); err != nil {
return err
}
if err := r.setParticipantRTCNode(participantId, rtcNode); err != nil {
return err
}
err = r.setParticipantSignalNode(participantId, r.currentNode.Id)
sink, err := r.GetRequestSink(participantId)
if err != nil {
return err
}
// sends a message to start session
return sink.WriteMessage(&livekit.StartSession{
RoomName: roomName,
ParticipantName: participantName,
})
}
func (r *RedisRouter) startParticipantRTC(roomName, participantId, participantName string) error {
// find the node where the room is hosted at
rtcNode, err := r.GetNodeForRoom(roomName)
if err != nil {
return err
}
if rtcNode != r.currentNode.Id {
logger.Errorw("called participant on incorrect node",
"rtcNode", rtcNode, "currentNode", r.currentNode.Id)
return ErrIncorrectRTCNode
}
// find signal node to send responses back
signalNode, err := r.getParticipantSignalNode(participantId)
if err != nil {
@@ -204,6 +203,15 @@ func (r *RedisRouter) Stop() {
r.cancel()
}
func (r *RedisRouter) setParticipantRTCNode(participantId, nodeId string) error {
r.cr.Expire(participantRTCKey(participantId))
err := r.rc.Set(r.ctx, participantRTCKey(participantId), nodeId, participantMappingTTL).Err()
if err != nil {
err = errors.Wrap(err, "could not set rtc node")
}
return err
}
func (r *RedisRouter) setParticipantSignalNode(participantId, nodeId string) error {
r.cr.Expire(participantSignalKey(participantId))
if err := r.rc.Set(r.ctx, participantSignalKey(participantId), nodeId, participantMappingTTL).Err(); err != nil {
@@ -213,9 +221,9 @@ func (r *RedisRouter) setParticipantSignalNode(participantId, nodeId string) err
}
func (r *RedisRouter) getOrCreateRedisSink(nodeId string, participantId string) *RedisSink {
r.lock.RLock()
r.lock.Lock()
defer r.lock.Unlock()
sink := r.redisSinks[participantId]
r.lock.RUnlock()
if sink != nil {
return sink
@@ -227,9 +235,7 @@ func (r *RedisRouter) getOrCreateRedisSink(nodeId string, participantId string)
delete(r.redisSinks, participantId)
r.lock.Unlock()
})
r.lock.Lock()
r.redisSinks[participantId] = sink
r.lock.Unlock()
return sink
}
@@ -283,8 +289,10 @@ func (r *RedisRouter) subscribeWorker() {
switch rmb := rm.Message.(type) {
case *livekit.RouterMessage_StartSession:
logger.Infow("received router startSession", "node", r.currentNode.Id,
"participant", pId)
// RTC session should start on this node
err = r.StartParticipant(rmb.StartSession.RoomName, pId, rmb.StartSession.ParticipantName)
err = r.startParticipantRTC(rmb.StartSession.RoomName, pId, rmb.StartSession.ParticipantName)
if err != nil {
logger.Errorw("could not start participant", "error", err)
}
pkg/routing/routingfakes/fake_router.go
+63 -139
View File
@@ -84,9 +84,9 @@ type FakeRouter struct {
result1 []*livekit.Node
result2 error
}
OnNewParticipantStub func(routing.ParticipantCallback)
onNewParticipantMutex sync.RWMutex
onNewParticipantArgsForCall []struct {
OnNewParticipantRTCStub func(routing.ParticipantCallback)
onNewParticipantRTCMutex sync.RWMutex
onNewParticipantRTCArgsForCall []struct {
arg1 routing.ParticipantCallback
}
RegisterNodeStub func() error
@@ -111,18 +111,6 @@ type FakeRouter struct {
setNodeForRoomReturnsOnCall map[int]struct {
result1 error
}
SetParticipantRTCNodeStub func(string, string) error
setParticipantRTCNodeMutex sync.RWMutex
setParticipantRTCNodeArgsForCall []struct {
arg1 string
arg2 string
}
setParticipantRTCNodeReturns struct {
result1 error
}
setParticipantRTCNodeReturnsOnCall map[int]struct {
result1 error
}
StartStub func() error
startMutex sync.RWMutex
startArgsForCall []struct {
@@ -133,17 +121,17 @@ type FakeRouter struct {
startReturnsOnCall map[int]struct {
result1 error
}
StartParticipantStub func(string, string, string) error
startParticipantMutex sync.RWMutex
startParticipantArgsForCall []struct {
StartParticipantSignalStub func(string, string, string) error
startParticipantSignalMutex sync.RWMutex
startParticipantSignalArgsForCall []struct {
arg1 string
arg2 string
arg3 string
}
startParticipantReturns struct {
startParticipantSignalReturns struct {
result1 error
}
startParticipantReturnsOnCall map[int]struct {
startParticipantSignalReturnsOnCall map[int]struct {
result1 error
}
StopStub func()
@@ -537,35 +525,35 @@ func (fake *FakeRouter) ListNodesReturnsOnCall(i int, result1 []*livekit.Node, r
}{result1, result2}
}
func (fake *FakeRouter) OnNewParticipant(arg1 routing.ParticipantCallback) {
fake.onNewParticipantMutex.Lock()
fake.onNewParticipantArgsForCall = append(fake.onNewParticipantArgsForCall, struct {
func (fake *FakeRouter) OnNewParticipantRTC(arg1 routing.ParticipantCallback) {
fake.onNewParticipantRTCMutex.Lock()
fake.onNewParticipantRTCArgsForCall = append(fake.onNewParticipantRTCArgsForCall, struct {
arg1 routing.ParticipantCallback
}{arg1})
stub := fake.OnNewParticipantStub
fake.recordInvocation("OnNewParticipant", []interface{}{arg1})
fake.onNewParticipantMutex.Unlock()
stub := fake.OnNewParticipantRTCStub
fake.recordInvocation("OnNewParticipantRTC", []interface{}{arg1})
fake.onNewParticipantRTCMutex.Unlock()
if stub != nil {
fake.OnNewParticipantStub(arg1)
fake.OnNewParticipantRTCStub(arg1)
}
}
func (fake *FakeRouter) OnNewParticipantCallCount() int {
fake.onNewParticipantMutex.RLock()
defer fake.onNewParticipantMutex.RUnlock()
return len(fake.onNewParticipantArgsForCall)
func (fake *FakeRouter) OnNewParticipantRTCCallCount() int {
fake.onNewParticipantRTCMutex.RLock()
defer fake.onNewParticipantRTCMutex.RUnlock()
return len(fake.onNewParticipantRTCArgsForCall)
}
func (fake *FakeRouter) OnNewParticipantCalls(stub func(routing.ParticipantCallback)) {
fake.onNewParticipantMutex.Lock()
defer fake.onNewParticipantMutex.Unlock()
fake.OnNewParticipantStub = stub
func (fake *FakeRouter) OnNewParticipantRTCCalls(stub func(routing.ParticipantCallback)) {
fake.onNewParticipantRTCMutex.Lock()
defer fake.onNewParticipantRTCMutex.Unlock()
fake.OnNewParticipantRTCStub = stub
}
func (fake *FakeRouter) OnNewParticipantArgsForCall(i int) routing.ParticipantCallback {
fake.onNewParticipantMutex.RLock()
defer fake.onNewParticipantMutex.RUnlock()
argsForCall := fake.onNewParticipantArgsForCall[i]
func (fake *FakeRouter) OnNewParticipantRTCArgsForCall(i int) routing.ParticipantCallback {
fake.onNewParticipantRTCMutex.RLock()
defer fake.onNewParticipantRTCMutex.RUnlock()
argsForCall := fake.onNewParticipantRTCArgsForCall[i]
return argsForCall.arg1
}
@@ -684,68 +672,6 @@ func (fake *FakeRouter) SetNodeForRoomReturnsOnCall(i int, result1 error) {
}{result1}
}
func (fake *FakeRouter) SetParticipantRTCNode(arg1 string, arg2 string) error {
fake.setParticipantRTCNodeMutex.Lock()
ret, specificReturn := fake.setParticipantRTCNodeReturnsOnCall[len(fake.setParticipantRTCNodeArgsForCall)]
fake.setParticipantRTCNodeArgsForCall = append(fake.setParticipantRTCNodeArgsForCall, struct {
arg1 string
arg2 string
}{arg1, arg2})
stub := fake.SetParticipantRTCNodeStub
fakeReturns := fake.setParticipantRTCNodeReturns
fake.recordInvocation("SetParticipantRTCNode", []interface{}{arg1, arg2})
fake.setParticipantRTCNodeMutex.Unlock()
if stub != nil {
return stub(arg1, arg2)
}
if specificReturn {
return ret.result1
}
return fakeReturns.result1
}
func (fake *FakeRouter) SetParticipantRTCNodeCallCount() int {
fake.setParticipantRTCNodeMutex.RLock()
defer fake.setParticipantRTCNodeMutex.RUnlock()
return len(fake.setParticipantRTCNodeArgsForCall)
}
func (fake *FakeRouter) SetParticipantRTCNodeCalls(stub func(string, string) error) {
fake.setParticipantRTCNodeMutex.Lock()
defer fake.setParticipantRTCNodeMutex.Unlock()
fake.SetParticipantRTCNodeStub = stub
}
func (fake *FakeRouter) SetParticipantRTCNodeArgsForCall(i int) (string, string) {
fake.setParticipantRTCNodeMutex.RLock()
defer fake.setParticipantRTCNodeMutex.RUnlock()
argsForCall := fake.setParticipantRTCNodeArgsForCall[i]
return argsForCall.arg1, argsForCall.arg2
}
func (fake *FakeRouter) SetParticipantRTCNodeReturns(result1 error) {
fake.setParticipantRTCNodeMutex.Lock()
defer fake.setParticipantRTCNodeMutex.Unlock()
fake.SetParticipantRTCNodeStub = nil
fake.setParticipantRTCNodeReturns = struct {
result1 error
}{result1}
}
func (fake *FakeRouter) SetParticipantRTCNodeReturnsOnCall(i int, result1 error) {
fake.setParticipantRTCNodeMutex.Lock()
defer fake.setParticipantRTCNodeMutex.Unlock()
fake.SetParticipantRTCNodeStub = nil
if fake.setParticipantRTCNodeReturnsOnCall == nil {
fake.setParticipantRTCNodeReturnsOnCall = make(map[int]struct {
result1 error
})
}
fake.setParticipantRTCNodeReturnsOnCall[i] = struct {
result1 error
}{result1}
}
func (fake *FakeRouter) Start() error {
fake.startMutex.Lock()
ret, specificReturn := fake.startReturnsOnCall[len(fake.startArgsForCall)]
@@ -799,18 +725,18 @@ func (fake *FakeRouter) StartReturnsOnCall(i int, result1 error) {
}{result1}
}
func (fake *FakeRouter) StartParticipant(arg1 string, arg2 string, arg3 string) error {
fake.startParticipantMutex.Lock()
ret, specificReturn := fake.startParticipantReturnsOnCall[len(fake.startParticipantArgsForCall)]
fake.startParticipantArgsForCall = append(fake.startParticipantArgsForCall, struct {
func (fake *FakeRouter) StartParticipantSignal(arg1 string, arg2 string, arg3 string) error {
fake.startParticipantSignalMutex.Lock()
ret, specificReturn := fake.startParticipantSignalReturnsOnCall[len(fake.startParticipantSignalArgsForCall)]
fake.startParticipantSignalArgsForCall = append(fake.startParticipantSignalArgsForCall, struct {
arg1 string
arg2 string
arg3 string
}{arg1, arg2, arg3})
stub := fake.StartParticipantStub
fakeReturns := fake.startParticipantReturns
fake.recordInvocation("StartParticipant", []interface{}{arg1, arg2, arg3})
fake.startParticipantMutex.Unlock()
stub := fake.StartParticipantSignalStub
fakeReturns := fake.startParticipantSignalReturns
fake.recordInvocation("StartParticipantSignal", []interface{}{arg1, arg2, arg3})
fake.startParticipantSignalMutex.Unlock()
if stub != nil {
return stub(arg1, arg2, arg3)
}
@@ -820,44 +746,44 @@ func (fake *FakeRouter) StartParticipant(arg1 string, arg2 string, arg3 string)
return fakeReturns.result1
}
func (fake *FakeRouter) StartParticipantCallCount() int {
fake.startParticipantMutex.RLock()
defer fake.startParticipantMutex.RUnlock()
return len(fake.startParticipantArgsForCall)
func (fake *FakeRouter) StartParticipantSignalCallCount() int {
fake.startParticipantSignalMutex.RLock()
defer fake.startParticipantSignalMutex.RUnlock()
return len(fake.startParticipantSignalArgsForCall)
}
func (fake *FakeRouter) StartParticipantCalls(stub func(string, string, string) error) {
fake.startParticipantMutex.Lock()
defer fake.startParticipantMutex.Unlock()
fake.StartParticipantStub = stub
func (fake *FakeRouter) StartParticipantSignalCalls(stub func(string, string, string) error) {
fake.startParticipantSignalMutex.Lock()
defer fake.startParticipantSignalMutex.Unlock()
fake.StartParticipantSignalStub = stub
}
func (fake *FakeRouter) StartParticipantArgsForCall(i int) (string, string, string) {
fake.startParticipantMutex.RLock()
defer fake.startParticipantMutex.RUnlock()
argsForCall := fake.startParticipantArgsForCall[i]
func (fake *FakeRouter) StartParticipantSignalArgsForCall(i int) (string, string, string) {
fake.startParticipantSignalMutex.RLock()
defer fake.startParticipantSignalMutex.RUnlock()
argsForCall := fake.startParticipantSignalArgsForCall[i]
return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
}
func (fake *FakeRouter) StartParticipantReturns(result1 error) {
fake.startParticipantMutex.Lock()
defer fake.startParticipantMutex.Unlock()
fake.StartParticipantStub = nil
fake.startParticipantReturns = struct {
func (fake *FakeRouter) StartParticipantSignalReturns(result1 error) {
fake.startParticipantSignalMutex.Lock()
defer fake.startParticipantSignalMutex.Unlock()
fake.StartParticipantSignalStub = nil
fake.startParticipantSignalReturns = struct {
result1 error
}{result1}
}
func (fake *FakeRouter) StartParticipantReturnsOnCall(i int, result1 error) {
fake.startParticipantMutex.Lock()
defer fake.startParticipantMutex.Unlock()
fake.StartParticipantStub = nil
if fake.startParticipantReturnsOnCall == nil {
fake.startParticipantReturnsOnCall = make(map[int]struct {
func (fake *FakeRouter) StartParticipantSignalReturnsOnCall(i int, result1 error) {
fake.startParticipantSignalMutex.Lock()
defer fake.startParticipantSignalMutex.Unlock()
fake.StartParticipantSignalStub = nil
if fake.startParticipantSignalReturnsOnCall == nil {
fake.startParticipantSignalReturnsOnCall = make(map[int]struct {
result1 error
})
}
fake.startParticipantReturnsOnCall[i] = struct {
fake.startParticipantSignalReturnsOnCall[i] = struct {
result1 error
}{result1}
}
@@ -954,18 +880,16 @@ func (fake *FakeRouter) Invocations() map[string][][]interface{} {
defer fake.getResponseSourceMutex.RUnlock()
fake.listNodesMutex.RLock()
defer fake.listNodesMutex.RUnlock()
fake.onNewParticipantMutex.RLock()
defer fake.onNewParticipantMutex.RUnlock()
fake.onNewParticipantRTCMutex.RLock()
defer fake.onNewParticipantRTCMutex.RUnlock()
fake.registerNodeMutex.RLock()
defer fake.registerNodeMutex.RUnlock()
fake.setNodeForRoomMutex.RLock()
defer fake.setNodeForRoomMutex.RUnlock()
fake.setParticipantRTCNodeMutex.RLock()
defer fake.setParticipantRTCNodeMutex.RUnlock()
fake.startMutex.RLock()
defer fake.startMutex.RUnlock()
fake.startParticipantMutex.RLock()
defer fake.startParticipantMutex.RUnlock()
fake.startParticipantSignalMutex.RLock()
defer fake.startParticipantSignalMutex.RUnlock()
fake.stopMutex.RLock()
defer fake.stopMutex.RUnlock()
fake.unregisterNodeMutex.RLock()
pkg/service/roommanager.go
-6
View File
@@ -150,12 +150,6 @@ func (r *RoomManager) StartSession(roomName, participantId, participantName stri
return
}
// register participant to be on this server
if err = r.router.SetParticipantRTCNode(participantId, r.currentNode.Id); err != nil {
logger.Errorw("could not set RTC node", "error", err)
return
}
// join room
if err := room.Join(participant); err != nil {
logger.Errorw("could not join room", "error", err)
pkg/service/rtcservice.go
+2 -1
View File
@@ -71,7 +71,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
}
participantId := utils.NewGuid(utils.ParticipantPrefix)
err = s.router.StartParticipant(roomName, participantId, pName)
err = s.router.StartParticipantSignal(roomName, participantId, pName)
if err != nil {
handleError(w, http.StatusInternalServerError, "could not start session: "+err.Error())
return
@@ -109,6 +109,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
"room", rm.Sid,
"roomName", rm.Name,
"name", pName,
"resSource", fmt.Sprintf("%p", resSource),
)
// handle responses
pkg/service/server.go
+1 -1
View File
@@ -62,7 +62,7 @@ func NewLivekitServer(conf *config.Config,
}
// hook up router to the RoomManager
router.OnNewParticipant(roomManager.StartSession)
router.OnNewParticipantRTC(roomManager.StartSession)
// clean up old rooms on startup
err = roomManager.Cleanup()
test/integration_helpers.go
+11 -6
View File
@@ -27,8 +27,13 @@ const (
testRoom = "mytestroom"
defaultServerPort = 7880
secondServerPort = 7881
nodeId1 = "integration-test-1"
nodeId2 = "integration-test-2"
nodeId1 = "node-1"
nodeId2 = "node-2"
connectTimeout = 10 * time.Second
// if there are deadlocks, it's helpful to set a short test timeout (i.e. go test -timeout=30s)
// let connection timeout happen
//connectTimeout = 5000 * time.Second
)
var (
@@ -82,7 +87,7 @@ func contextWithCreateRoomToken() context.Context {
func waitForServerToStart(s *service.LivekitServer) {
// wait till ready
ctx, _ := context.WithTimeout(context.Background(), 10*time.Second)
ctx, _ := context.WithTimeout(context.Background(), connectTimeout)
for {
select {
case <-ctx.Done():
@@ -96,7 +101,7 @@ func waitForServerToStart(s *service.LivekitServer) {
}
func withTimeout(t *testing.T, description string, f func() bool) {
ctx, _ := context.WithTimeout(context.Background(), 10*time.Second)
ctx, _ := context.WithTimeout(context.Background(), connectTimeout)
for {
select {
case <-ctx.Done():
@@ -117,7 +122,7 @@ func waitUntilConnected(t *testing.T, clients ...*client.RTCClient) {
go func() {
defer wg.Done()
if !assert.NoError(t, c.WaitUntilConnected()) {
t.Fatal("one or more clients could not connect")
t.Fatal("client could not connect", c.ID())
}
}()
}
@@ -170,7 +175,7 @@ func createMultiNodeServer(nodeId string, port uint32) *service.LivekitServer {
panic(err)
}
router := routing.NewRedisRouter(currentNode, rc, false)
router := routing.NewRedisRouter(currentNode, rc)
roomStore := service.NewRedisRoomStore(rc)
s, err := service.InitializeServer(conf, &StaticKeyProvider{}, roomStore, router, currentNode, &routing.RandomSelector{})
if err != nil {