From 25d476b8ac13557db69c80310b68d294c1f18d7c Mon Sep 17 00:00:00 2001 From: David Zhao Date: Wed, 27 Jan 2021 01:16:08 -0800 Subject: [PATCH] fixed multi-node routing, correctly set rtc and signal nodes --- cmd/server/main.go | 2 +- pkg/routing/errors.go | 1 + pkg/routing/interfaces.go | 8 +- pkg/routing/localrouter.go | 18 +-- pkg/routing/redisrouter.go | 88 ++++++----- pkg/routing/routingfakes/fake_router.go | 202 ++++++++---------------- pkg/service/roommanager.go | 6 - pkg/service/rtcservice.go | 3 +- pkg/service/server.go | 2 +- test/integration_helpers.go | 17 +- 10 files changed, 136 insertions(+), 211 deletions(-) diff --git a/cmd/server/main.go b/cmd/server/main.go index 444be014e..9ea58d4ee 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -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 diff --git a/pkg/routing/errors.go b/pkg/routing/errors.go index ea4b5d325..35e45a654 100644 --- a/pkg/routing/errors.go +++ b/pkg/routing/errors.go @@ -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") ) diff --git a/pkg/routing/interfaces.go b/pkg/routing/interfaces.go index ec34e9e54..966dcddc8 100644 --- a/pkg/routing/interfaces.go +++ b/pkg/routing/interfaces.go @@ -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() } diff --git a/pkg/routing/localrouter.go b/pkg/routing/localrouter.go index 68bd9835d..482a932ce 100644 --- a/pkg/routing/localrouter.go +++ b/pkg/routing/localrouter.go @@ -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 } diff --git a/pkg/routing/redisrouter.go b/pkg/routing/redisrouter.go index f1053d3ed..78a1db6b6 100644 --- a/pkg/routing/redisrouter.go +++ b/pkg/routing/redisrouter.go @@ -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) } diff --git a/pkg/routing/routingfakes/fake_router.go b/pkg/routing/routingfakes/fake_router.go index b81891c68..c81c9926f 100644 --- a/pkg/routing/routingfakes/fake_router.go +++ b/pkg/routing/routingfakes/fake_router.go @@ -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() diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index a6dafbd4d..725f2efa9 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -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) diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index 62eb11db4..efb410103 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -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 diff --git a/pkg/service/server.go b/pkg/service/server.go index d3777f690..9fc515cb9 100644 --- a/pkg/service/server.go +++ b/pkg/service/server.go @@ -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() diff --git a/test/integration_helpers.go b/test/integration_helpers.go index d123edf67..3da5fb1af 100644 --- a/test/integration_helpers.go +++ b/test/integration_helpers.go @@ -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 {