diff --git a/pkg/routing/errors.go b/pkg/routing/errors.go index 5737ec6ac..2a6c82db9 100644 --- a/pkg/routing/errors.go +++ b/pkg/routing/errors.go @@ -7,6 +7,7 @@ var ( 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") + ErrNodeNotFound = errors.New("could not locate the node") errInvalidRouterMessage = errors.New("invalid router message") ErrChannelClosed = errors.New("channel closed") ErrChannelFull = errors.New("channel is full") diff --git a/pkg/routing/redisrouter.go b/pkg/routing/redisrouter.go index 4b6b78fd8..e53abbc74 100644 --- a/pkg/routing/redisrouter.go +++ b/pkg/routing/redisrouter.go @@ -273,11 +273,19 @@ func (r *RedisRouter) setParticipantSignalNode(connectionId, nodeId string) erro } func (r *RedisRouter) getParticipantRTCNode(participantKey string) (string, error) { - return r.rc.Get(r.ctx, participantRTCKey(participantKey)).Result() + val, err := r.rc.Get(r.ctx, participantRTCKey(participantKey)).Result() + if err == redis.Nil { + err = ErrNodeNotFound + } + return val, err } func (r *RedisRouter) getParticipantSignalNode(connectionId string) (nodeId string, err error) { - return r.rc.Get(r.ctx, participantSignalKey(connectionId)).Result() + val, err := r.rc.Get(r.ctx, participantSignalKey(connectionId)).Result() + if err == redis.Nil { + err = ErrNodeNotFound + } + return val, err } // update node stats and cleanup diff --git a/pkg/service/redisroomstore.go b/pkg/service/redisroomstore.go index a3bac34c3..32b80d450 100644 --- a/pkg/service/redisroomstore.go +++ b/pkg/service/redisroomstore.go @@ -12,13 +12,13 @@ import ( ) const ( - // hash of room_name => Room proto + // RoomsKey is hash of room_name => Room proto RoomsKey = "rooms" - // hash of room_id => room name + // RoomIdMap is hash of room_id => room name RoomIdMap = "room_id_map" - // hash of participant_name => ParticipantInfo + // RoomParticipantsPrefix is hash of participant_name => ParticipantInfo // a key for each room, with expiration RoomParticipantsPrefix = "room_participants:" diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index bf205ef35..49a92d7ae 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -395,6 +395,11 @@ func (r *RoomManager) handleRTCMessage(roomName, identity string, msg *livekit.R if rm.UpdateParticipant.Permission != nil { participant.SetPermission(rm.UpdateParticipant.Permission) } + case *livekit.RTCNodeMessage_DeleteRoom: + for _, p := range room.GetParticipants() { + _ = p.Close() + } + room.Close() } } diff --git a/pkg/service/roomservice.go b/pkg/service/roomservice.go index 4cad80ffa..9ee5019e9 100644 --- a/pkg/service/roomservice.go +++ b/pkg/service/roomservice.go @@ -54,17 +54,38 @@ func (s *RoomService) ListRooms(ctx context.Context, req *livekit.ListRoomsReque return } -func (s *RoomService) DeleteRoom(ctx context.Context, req *livekit.DeleteRoomRequest) (res *livekit.DeleteRoomResponse, err error) { - if err = EnsureCreatePermission(ctx); err != nil { +func (s *RoomService) DeleteRoom(ctx context.Context, req *livekit.DeleteRoomRequest) (*livekit.DeleteRoomResponse, error) { + if err := EnsureCreatePermission(ctx); err != nil { return nil, twirpAuthError(err) } - err = s.roomManager.DeleteRoom(req.Room) + // if the room is currently active, RTC node needs to disconnect clients + // here we are using any user's identity, due to how it works with routing + participants, err := s.roomManager.roomStore.ListParticipants(req.Room) if err != nil { - err = twirp.WrapError(twirp.InternalError("could not delete room"), err) - return + return nil, err } - res = &livekit.DeleteRoomResponse{} - return + + if len(participants) > 0 { + rtcSink, err := s.roomManager.router.CreateRTCSink(req.Room, participants[0].Identity) + if err != nil { + return nil, err + } + + defer rtcSink.Close() + _ = rtcSink.WriteMessage(&livekit.RTCNodeMessage{ + Message: &livekit.RTCNodeMessage_DeleteRoom{ + DeleteRoom: req, + }, + }) + } else { + // if a room hasn't started, delete locally + if err = s.roomManager.DeleteRoom(req.Room); err != nil { + err = twirp.WrapError(twirp.InternalError("could not delete room"), err) + return nil, err + } + } + + return &livekit.DeleteRoomResponse{}, nil } func (s *RoomService) ListParticipants(ctx context.Context, req *livekit.ListParticipantsRequest) (res *livekit.ListParticipantsResponse, err error) {