diff --git a/go.mod b/go.mod index a903f567..0cabb6cb 100644 --- a/go.mod +++ b/go.mod @@ -15,7 +15,7 @@ require ( github.com/jordic/lti v0.0.0-20160211051708-2c756eacbab9 github.com/livekit/protocol v1.21.0 github.com/livekit/server-sdk-go/v2 v2.2.1 - github.com/mynaparrot/plugnmeet-protocol v0.0.0-20240914113712-e6e9efae7d51 + github.com/mynaparrot/plugnmeet-protocol v0.0.0-20240916153048-d72bf01c12fc github.com/nats-io/jwt/v2 v2.6.0 github.com/nats-io/nats.go v1.37.0 github.com/nats-io/nkeys v0.4.7 diff --git a/go.sum b/go.sum index 281dc5de..9999d2f6 100644 --- a/go.sum +++ b/go.sum @@ -128,8 +128,8 @@ github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWE github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/mattn/go-runewidth v0.0.16 h1:E5ScNMtiwvlvB5paMFdw9p4kSQzbXFikJ5SQO6TULQc= github.com/mattn/go-runewidth v0.0.16/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w= -github.com/mynaparrot/plugnmeet-protocol v0.0.0-20240914113712-e6e9efae7d51 h1:M/3/IyT3ps5OoO+c8GlnVDsvuBQJ5jWMvK6blA5kOdY= -github.com/mynaparrot/plugnmeet-protocol v0.0.0-20240914113712-e6e9efae7d51/go.mod h1:oXYXVoAr5AQAO785GGRZdaF2Pn1t8PZsjoIs3cGcSJY= +github.com/mynaparrot/plugnmeet-protocol v0.0.0-20240916153048-d72bf01c12fc h1:V75DG+KfPXg2goGSs7LXYPPxdNrqxnEVs5oZWGJPcek= +github.com/mynaparrot/plugnmeet-protocol v0.0.0-20240916153048-d72bf01c12fc/go.mod h1:oXYXVoAr5AQAO785GGRZdaF2Pn1t8PZsjoIs3cGcSJY= github.com/nats-io/jwt/v2 v2.6.0 h1:yXoBTdEotZw3NujMT+Nnu1UPNlFWdKQ3d0JJF/+pJag= github.com/nats-io/jwt/v2 v2.6.0/go.mod h1:ZdWS1nZa6WMZfFwwgpEaqBV8EPGVgOTDHN/wTbz0Y5A= github.com/nats-io/nats.go v1.37.0 h1:07rauXbVnnJvv1gfIyghFEo6lUcYRY0WXc3x7x0vUxE= diff --git a/pkg/models/bkroom_end.go b/pkg/models/bkroom_end.go index 2bd5fb91..f5deb528 100644 --- a/pkg/models/bkroom_end.go +++ b/pkg/models/bkroom_end.go @@ -1,41 +1,56 @@ package models import ( + "errors" "github.com/mynaparrot/plugnmeet-protocol/plugnmeet" + log "github.com/sirupsen/logrus" ) func (m *BreakoutRoomModel) EndBreakoutRoom(r *plugnmeet.EndBreakoutRoomReq) error { - _, err := m.fetchBreakoutRoom(r.RoomId, r.BreakoutRoomId) + rm, err := m.natsService.GetBreakoutRoom(r.RoomId, r.BreakoutRoomId) if err != nil { return err } - _, _ = m.rm.EndRoom(&plugnmeet.RoomEndReq{RoomId: r.BreakoutRoomId}) - - _ = m.natsService.DeleteBreakoutRoom(r.RoomId, r.BreakoutRoomId) - m.performPostHookTask(r.RoomId) + if rm == nil { + return errors.New("room not found") + } + m.proceedToEndBkRoom(r.BreakoutRoomId, r.RoomId) return nil } func (m *BreakoutRoomModel) EndAllBreakoutRoomsByParentRoomId(parentRoomId string) error { - rooms, err := m.fetchBreakoutRooms(parentRoomId) + ids, err := m.natsService.GetBreakoutRoomIdsByParentRoomId(parentRoomId) if err != nil { return err } - if rooms == nil || len(rooms) == 0 { + if ids == nil || len(ids) == 0 { return m.updateParentRoomMetadata(parentRoomId) } - for _, r := range rooms { - _ = m.EndBreakoutRoom(&plugnmeet.EndBreakoutRoomReq{ - BreakoutRoomId: r.Id, - RoomId: parentRoomId, - }) + for _, i := range ids { + m.proceedToEndBkRoom(i, parentRoomId) } return nil } -func (m *BreakoutRoomModel) performPostHookTask(roomId string) { +func (m *BreakoutRoomModel) proceedToEndBkRoom(bkRoomId, parentRoomId string) { + ok, msg := m.rm.EndRoom(&plugnmeet.RoomEndReq{RoomId: bkRoomId}) + if !ok { + log.Errorln(msg) + } + + err := m.natsService.DeleteBreakoutRoom(parentRoomId, bkRoomId) + if err != nil { + log.Errorln(err) + } + + m.onAfterBkRoomEnded(parentRoomId) + // notify to the room for updating list + _ = m.natsService.BroadcastSystemEventToRoom(plugnmeet.NatsMsgServerToClientEvents_BREAKOUT_ROOM_ENDED, parentRoomId, bkRoomId, nil) +} + +func (m *BreakoutRoomModel) onAfterBkRoomEnded(roomId string) { if c, err := m.natsService.CountBreakoutRooms(roomId); err == nil && c == 0 { // no room left so, delete breakoutRoomKey key for this room m.natsService.DeleteAllBreakoutRoomsByParentRoomId(roomId) @@ -78,7 +93,7 @@ func (m *BreakoutRoomModel) PostTaskAfterRoomEndWebhook(roomId, metadata string) if meta.IsBreakoutRoom { _ = m.natsService.DeleteBreakoutRoom(meta.ParentRoomId, roomId) - m.performPostHookTask(meta.ParentRoomId) + m.onAfterBkRoomEnded(meta.ParentRoomId) } else { err = m.EndAllBreakoutRoomsByParentRoomId(roomId) if err != nil { diff --git a/pkg/services/nats/bk_room.go b/pkg/services/nats/bk_room.go index d53cbc89..f8ac5f69 100644 --- a/pkg/services/nats/bk_room.go +++ b/pkg/services/nats/bk_room.go @@ -33,8 +33,7 @@ func (s *NatsService) DeleteBreakoutRoom(parentRoomId, bkRoomId string) error { return err } - _ = kv.Purge(s.ctx, bkRoomId) - return nil + return kv.Purge(s.ctx, bkRoomId) } func (s *NatsService) GetBreakoutRoom(parentRoomId, bkRoomId string) ([]byte, error) { @@ -71,12 +70,14 @@ func (s *NatsService) CountBreakoutRooms(parentRoomId string) (int64, error) { } keys, err := kv.ListKeys(s.ctx) - defer keys.Stop() if err != nil { return 0, err } - - return int64(len(keys.Keys())), nil + var count int64 + for range keys.Keys() { + count++ + } + return count, nil } func (s *NatsService) GetAllBreakoutRoomsByParentRoomId(parentRoomId string) (map[string][]byte, error) { @@ -106,3 +107,25 @@ func (s *NatsService) GetAllBreakoutRoomsByParentRoomId(parentRoomId string) (ma func (s *NatsService) DeleteAllBreakoutRoomsByParentRoomId(parentRoomId string) { _ = s.js.DeleteKeyValue(s.ctx, fmt.Sprintf(breakoutRoomBucket, parentRoomId)) } + +func (s *NatsService) GetBreakoutRoomIdsByParentRoomId(parentRoomId string) ([]string, error) { + kv, err := s.js.KeyValue(s.ctx, fmt.Sprintf(breakoutRoomBucket, parentRoomId)) + switch { + case errors.Is(err, jetstream.ErrBucketNotFound): + return nil, nil + case err != nil: + return nil, err + } + + keys, err := kv.ListKeys(s.ctx) + if err != nil { + return nil, err + } + + var ids []string + for k := range keys.Keys() { + ids = append(ids, k) + } + + return ids, nil +} diff --git a/pkg/services/nats/etherpad.go b/pkg/services/nats/etherpad.go index 67c1c926..6391a151 100644 --- a/pkg/services/nats/etherpad.go +++ b/pkg/services/nats/etherpad.go @@ -37,12 +37,16 @@ func (s *NatsService) GetEtherpadActiveRoomsNum(nodeId string) (int64, error) { } keys, err := kv.ListKeys(s.ctx) - defer keys.Stop() if err != nil { return 0, err } - return int64(len(keys.Keys())), nil + var count int64 + for range keys.Keys() { + count++ + } + + return count, nil } func (s *NatsService) RemoveRoomFromEtherpad(nodeId, roomId string) error {