diff --git a/magefile.go b/magefile.go index 9d450d847..ec6e2df03 100644 --- a/magefile.go +++ b/magefile.go @@ -157,7 +157,7 @@ func Test() error { // run all thests including integration func TestAll() error { mg.Deps(Proto) - cmd := exec.Command("go", "test", "./...", "-count=1") + cmd := exec.Command("go", "test", "./...", "-count=1", "-timeout=1m") connectStd(cmd) return cmd.Run() } diff --git a/pkg/routing/errors.go b/pkg/routing/errors.go index 1f040d878..ea4b5d325 100644 --- a/pkg/routing/errors.go +++ b/pkg/routing/errors.go @@ -6,7 +6,6 @@ var ( ErrNotFound = errors.New("could not find object") ErrHandlerNotDefined = errors.New("handler not defined") ErrNoAvailableNodes = errors.New("could not find any available nodes") - ErrIncorrectNodeForRoom = errors.New("incorrect node for the current room") errInvalidRouterMessage = errors.New("invalid router message") ErrChannelClosed = errors.New("channel closed") ) diff --git a/pkg/routing/redisrouter.go b/pkg/routing/redisrouter.go index 83d1b78c6..f1053d3ed 100644 --- a/pkg/routing/redisrouter.go +++ b/pkg/routing/redisrouter.go @@ -161,12 +161,7 @@ func (r *RedisRouter) StartParticipant(roomName, participantId, participantName return err } - // StartParticipant should only be called on the current node - if rtcNode != r.currentNode.Id { - return ErrIncorrectNodeForRoom - } - - if r.useLocal { + if r.useLocal && rtcNode == r.currentNode.Id { return r.LocalRouter.StartParticipant(roomName, participantId, participantId) } @@ -282,7 +277,6 @@ func (r *RedisRouter) subscribeWorker() { continue } - logger.Debugw("received redis message", "message", msg, "node", r.currentNode.Id) rm := livekit.RouterMessage{} err = proto.Unmarshal([]byte(msg.Payload), &rm) pId := rm.ParticipantId diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index de6903393..5381ffe2f 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -109,7 +109,7 @@ func NewParticipant(participantId, name string, pc types.PeerConnection, rs rout ci := c.ToJSON() // write candidate - logger.Debugw("sending ice candidates") + //logger.Debugw("sending ice candidates") err := rs.WriteMessage(&livekit.SignalResponse{ Message: &livekit.SignalResponse_Trickle{ Trickle: ToProtoTrickle(ci), diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index fc280c9a3..a6dafbd4d 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -54,7 +54,8 @@ func (r *RoomManager) CreateRoom(req *livekit.CreateRoomRequest) (*livekit.Room, return nil, err } - if req.NodeId == "" { + nodeId := req.NodeId + if nodeId == "" { // select a node for room nodes, err := r.router.ListNodes() if err != nil { @@ -65,10 +66,11 @@ func (r *RoomManager) CreateRoom(req *livekit.CreateRoomRequest) (*livekit.Room, if err != nil { return nil, err } - req.NodeId = node.Id + nodeId = node.Id } - if err := r.router.SetNodeForRoom(req.Name, req.NodeId); err != nil { + logger.Debugw("selected node for room", "room", rm.Name, "node", nodeId) + if err := r.router.SetNodeForRoom(req.Name, nodeId); err != nil { return nil, err } @@ -131,6 +133,7 @@ func (r *RoomManager) StartSession(roomName, participantId, participantName stri logger.Debugw("starting RTC session", "room", roomName, + "node", r.currentNode.Id, "participant", participantName, "num_participants", len(room.GetParticipants()), ) @@ -201,7 +204,6 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.Partici }() defer rtc.Recover() - logger.Debugw("starting RTC session", "node", r.currentNode.Id) for { select { case <-time.After(time.Millisecond * 100): diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index 4c636221f..62eb11db4 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -70,6 +70,30 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } + participantId := utils.NewGuid(utils.ParticipantPrefix) + err = s.router.StartParticipant(roomName, participantId, pName) + if err != nil { + handleError(w, http.StatusInternalServerError, "could not start session: "+err.Error()) + return + } + reqSink, err := s.router.GetRequestSink(participantId) + if err != nil { + handleError(w, http.StatusInternalServerError, "could not get request sink"+err.Error()) + return + } + done := make(chan bool, 1) + defer func() { + logger.Infow("WS connection closed", "participant", pName) + reqSink.Close() + close(done) + }() + + resSource, err := s.router.GetResponseSource(participantId) + if err != nil { + handleError(w, http.StatusInternalServerError, "could not get response source"+err.Error()) + return + } + // upgrade only once the basics are good to go conn, err := s.upgrader.Upgrade(w, r, nil) if err != nil { @@ -79,39 +103,14 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { handleError(w, http.StatusInternalServerError, err.Error()) return } - sigConn := NewWSSignalConnection(conn) - participantId := utils.NewGuid(utils.ParticipantPrefix) - err = s.router.StartParticipant(roomName, participantId, pName) - if err != nil { - handleError(w, http.StatusInternalServerError, "could not start session: "+err.Error()) - return - } - logger.Infow("new client connected", "room", rm.Sid, "roomName", rm.Name, "name", pName, ) - reqSink, err := s.router.GetRequestSink(participantId) - if err != nil { - handleError(w, http.StatusInternalServerError, "could not get request sink"+err.Error()) - return - } - resSource, err := s.router.GetResponseSource(participantId) - if err != nil { - handleError(w, http.StatusInternalServerError, "could not get response source"+err.Error()) - return - } - done := make(chan bool, 1) - defer func() { - logger.Infow("WS connection closed", "participant", pName) - reqSink.Close() - close(done) - }() - // handle responses go func() { for { diff --git a/test/integration_helpers.go b/test/integration_helpers.go index 53663190e..d123edf67 100644 --- a/test/integration_helpers.go +++ b/test/integration_helpers.go @@ -115,10 +115,10 @@ func waitUntilConnected(t *testing.T, clients ...*client.RTCClient) { c := clients[i] wg.Add(1) go func() { + defer wg.Done() if !assert.NoError(t, c.WaitUntilConnected()) { t.Fatal("one or more clients could not connect") } - wg.Done() }() } wg.Wait() diff --git a/test/multinode_test.go b/test/multinode_test.go index 1b62fed16..faef3e4f7 100644 --- a/test/multinode_test.go +++ b/test/multinode_test.go @@ -31,13 +31,15 @@ func TestMultiNodeRouting(t *testing.T) { // one node connecting to node 1, and another connecting to node 2 c1 := createRTCClient("c1", defaultServerPort) - c2 := createRTCClient("c2", defaultServerPort) + c2 := createRTCClient("c2", secondServerPort) waitUntilConnected(t, c1, c2) // c1 publishing, and c2 receiving t1, err := c1.AddStaticTrack("audio/opus", "audio", "webcam") assert.NoError(t, err) - defer t1.Stop() + if t1 != nil { + defer t1.Stop() + } withTimeout(t, "c2 should receive one track", func() bool { if len(c2.SubscribedTracks()) == 0 { diff --git a/test/scenarios.go b/test/scenarios.go new file mode 100644 index 000000000..554b272f2 --- /dev/null +++ b/test/scenarios.go @@ -0,0 +1,10 @@ +package test + +import ( + "testing" +) + +// a scenario with lots of clients connecting, publishing, and leaving at random periods +func scenarioRandom(t *testing.T) { + +}