refact: make seperation of websocket channel (#52)

seperated websocket channels based on type

internal/controllers/websocket_service.go

@@ -1,7 +1,6 @@
 package controllers
 
 import (
-	"context"
 	"github.com/antoniodipinto/ikisocket"
 	"github.com/goccy/go-json"
 	"github.com/gofiber/fiber/v2"
@@ -119,14 +118,7 @@ func SetupSocketListeners() {
 			msg.IsAdmin = isAdmin.(bool)
 		}
 
-		marshal, err := json.Marshal(msg)
-		if err != nil {
-			log.Errorln(err)
-			return
-		}
-
-		ctx := context.Background()
-		config.AppCnf.RDS.Publish(ctx, "plug-n-meet-websocket", marshal)
+		models.DistributeWebsocketMsgToRedisChannel(&msg)
 	})
 
 	// On disconnect event
@@ -151,29 +143,9 @@ func SetupSocketListeners() {
 	//})
 }
 
-// SubscribeToWebsocketChannel will delivery message to websocket
+// SubscribeToWebsocketChannel will subscribe to all websocket channels
 func SubscribeToWebsocketChannel() {
-	ctx := context.Background()
-	pubsub := config.AppCnf.RDS.Subscribe(ctx, "plug-n-meet-websocket")
-	defer pubsub.Close()
-
-	_, err := pubsub.Receive(ctx)
-	if err != nil {
-		log.Fatalln(err)
-	}
-
-	m := models.NewWebsocketService()
-	ch := pubsub.Channel()
-	for msg := range ch {
-		res := new(models.WebsocketRedisMsg)
-		err = json.Unmarshal([]byte(msg.Payload), res)
-		if err != nil {
-			log.Errorln(err)
-		}
-		if res.Type == "sendMsg" {
-			m.HandleDataMessages(res.Payload, res.RoomId, res.IsAdmin)
-		} else if res.Type == "deleteRoom" {
-			config.AppCnf.DeleteChatRoom(res.RoomId)
-		}
-	}
+	go models.SubscribeToUserWebsocketChannel()
+	go models.SubscribeToWhiteboardWebsocketChannel()
+	go models.SubscribeToSystemWebsocketChannel()
 }

internal/models/breakout_room.go

@@ -384,19 +384,14 @@ func (m *breakoutRoom) broadcastNotification(roomId, fromUserId, toUserId, broad
 		payload.To = toUserId
 	}
 
-	msg := WebsocketRedisMsg{
+	msg := &WebsocketRedisMsg{
 		Type:    "sendMsg",
 		Payload: &payload,
 		RoomId:  roomId,
 		IsAdmin: isAdmin,
 	}
+	DistributeWebsocketMsgToRedisChannel(msg)
 
-	marshal, err := json.Marshal(msg)
-	if err != nil {
-		return err
-	}
-
-	m.rc.Publish(m.ctx, "plug-n-meet-websocket", marshal)
 	return nil
 }
 

internal/models/polls.go

@@ -269,19 +269,14 @@ func (m *newPollsModel) broadcastNotification(roomId, userId, pollId, mType stri
 		},
 	}
 
-	msg := WebsocketRedisMsg{
+	msg := &WebsocketRedisMsg{
 		Type:    "sendMsg",
 		Payload: &payload,
 		RoomId:  roomId,
 		IsAdmin: isAdmin,
 	}
+	DistributeWebsocketMsgToRedisChannel(msg)
 
-	marshal, err := json.Marshal(msg)
-	if err != nil {
-		return err
-	}
-
-	m.rc.Publish(m.ctx, "plug-n-meet-websocket", marshal)
 	return nil
 }
 

internal/models/webhook.go

@@ -128,7 +128,7 @@ func (w *webhookEvent) roomFinished() int64 {
 	}
 	marshal, err := json.Marshal(msg)
 	if err == nil {
-		config.AppCnf.RDS.Publish(context.Background(), "plug-n-meet-websocket", marshal)
+		config.AppCnf.RDS.Publish(context.Background(), "plug-n-meet-user-websocket", marshal)
 	}
 
 	// notify to clean room from room duration map

internal/models/websocket_channels.go

@@ -0,0 +1,99 @@
+package models
+
+import (
+	"context"
+	"github.com/goccy/go-json"
+	"github.com/mynaparrot/plugNmeet/internal/config"
+	log "github.com/sirupsen/logrus"
+)
+
+func DistributeWebsocketMsgToRedisChannel(msg *WebsocketRedisMsg) {
+	ctx := context.Background()
+	marshal, err := json.Marshal(msg)
+	if err != nil {
+		log.Errorln(err)
+		return
+	}
+
+	switch msg.Payload.Type {
+	case "USER":
+		config.AppCnf.RDS.Publish(ctx, "plug-n-meet-user-websocket", marshal)
+	case "WHITEBOARD":
+		config.AppCnf.RDS.Publish(ctx, "plug-n-meet-whiteboard-websocket", marshal)
+	case "SYSTEM":
+		config.AppCnf.RDS.Publish(ctx, "plug-n-meet-system-websocket", marshal)
+	}
+}
+
+// SubscribeToUserWebsocketChannel will delivery message to user websocket
+func SubscribeToUserWebsocketChannel() {
+	ctx := context.Background()
+	pubsub := config.AppCnf.RDS.Subscribe(ctx, "plug-n-meet-user-websocket")
+	defer pubsub.Close()
+
+	_, err := pubsub.Receive(ctx)
+	if err != nil {
+		log.Fatalln(err)
+	}
+
+	m := NewWebsocketService()
+	ch := pubsub.Channel()
+	for msg := range ch {
+		res := new(WebsocketRedisMsg)
+		err = json.Unmarshal([]byte(msg.Payload), res)
+		if err != nil {
+			log.Errorln(err)
+		}
+		if res.Type == "sendMsg" {
+			m.HandleDataMessages(res.Payload, res.RoomId, res.IsAdmin)
+		} else if res.Type == "deleteRoom" {
+			config.AppCnf.DeleteChatRoom(res.RoomId)
+		}
+	}
+}
+
+// SubscribeToWhiteboardWebsocketChannel will delivery message to whiteboard websocket
+func SubscribeToWhiteboardWebsocketChannel() {
+	ctx := context.Background()
+	pubsub := config.AppCnf.RDS.Subscribe(ctx, "plug-n-meet-whiteboard-websocket")
+	defer pubsub.Close()
+
+	_, err := pubsub.Receive(ctx)
+	if err != nil {
+		log.Fatalln(err)
+	}
+
+	m := NewWebsocketService()
+	ch := pubsub.Channel()
+	for msg := range ch {
+		res := new(WebsocketRedisMsg)
+		err = json.Unmarshal([]byte(msg.Payload), res)
+		if err != nil {
+			log.Errorln(err)
+		}
+		m.HandleDataMessages(res.Payload, res.RoomId, res.IsAdmin)
+	}
+}
+
+// SubscribeToSystemWebsocketChannel will delivery message to websocket
+func SubscribeToSystemWebsocketChannel() {
+	ctx := context.Background()
+	pubsub := config.AppCnf.RDS.Subscribe(ctx, "plug-n-meet-system-websocket")
+	defer pubsub.Close()
+
+	_, err := pubsub.Receive(ctx)
+	if err != nil {
+		log.Fatalln(err)
+	}
+
+	m := NewWebsocketService()
+	ch := pubsub.Channel()
+	for msg := range ch {
+		res := new(WebsocketRedisMsg)
+		err = json.Unmarshal([]byte(msg.Payload), res)
+		if err != nil {
+			log.Errorln(err)
+		}
+		m.HandleDataMessages(res.Payload, res.RoomId, res.IsAdmin)
+	}
+}