调整接口的逻辑

pkg/websocket/socket_new.go

@@ -3,6 +3,8 @@ package websocket
 import (
 	"context"
 	"fmt"
+	"github.com/ayflying/utility_go/pkg/aycache"
+	"github.com/ayflying/utility_go/tools"
 	"github.com/gogf/gf/v2/container/gmap"
 	"github.com/gogf/gf/v2/encoding/gjson"
 	"github.com/gogf/gf/v2/frame/g"
@@ -10,10 +12,10 @@ import (
 	"github.com/gogf/gf/v2/os/gctx"
 	"github.com/gogf/gf/v2/os/glog"
 	"github.com/gogf/gf/v2/util/guid"
-	"github.com/google/uuid"
 	"github.com/gorilla/websocket"
 	"google.golang.org/protobuf/proto"
 	"sync"
+	"time"
 )
 
 type SocketV1 struct {
@@ -23,17 +25,18 @@ type SocketV1 struct {
 var (
 	//ctx = gctx.New()
 	//Conn map[uuid.UUID]*WebsocketData
-	lock sync.Mutex
-
-	m = gmap.NewHashMap(true)
+	lock  sync.Mutex
+	cache = aycache.New("redis")
+	m     = gmap.NewHashMap(true)
 )
 
 type WebsocketData struct {
-	Ws     *websocket.Conn
-	Uuid   string
-	Uid    int64
-	Ctx    context.Context
-	RoomId int
+	Ws     *websocket.Conn `json:"ws" dc:"websocket连接池"`
+	Uuid   string          `json:"uuid" dc:"用户唯一标识"`
+	Uid    int64           `json:"uid" dc:"用户编号"`
+	Groups []string        `json:"groups" dc:"群组"`
+	Ctx    context.Context `json:"ctx" dc:""`
+	RoomId int             `json:"roomId" dc:"房间编号"`
 }
 
 func NewV1() *SocketV1 {
@@ -81,17 +84,17 @@ func (s *SocketV1) Load(serv *ghttp.Server, prefix string) {
 //	@Description:
 //	@receiver s
 //	@param conn
-func (s *SocketV1) OnConnect(ctx context.Context, conn *websocket.Conn) {
-
-	defer conn.Close()
+func (s *SocketV1) OnConnect(ctx context.Context, ws *websocket.Conn) {
 	id := guid.S()
-	ip := conn.RemoteAddr().String()
-	data := &WebsocketData{
-		Uuid: id,
-		Ws:   conn,
-		Ctx:  ctx,
+	ip := ws.RemoteAddr().String()
+	conn := &WebsocketData{
+		Uuid:   id,
+		Ws:     ws,
+		Ctx:    ctx,
+		Groups: make([]string, 0),
+		RoomId: -1,
 	}
-	m.Set(id, data)
+	m.Set(id, conn)
 
 	//defer delete(Conn, id)
 
@@ -101,22 +104,23 @@ func (s *SocketV1) OnConnect(ctx context.Context, conn *websocket.Conn) {
 
 	//用户登录钩子执行
 	for _, connect := range OnConnectHandlers {
-		connect(data)
+		connect(conn)
 	}
 
 	for {
 		//进入当前连接线程拥堵
-		msgType, msg, err := conn.ReadMessage()
+		msgType, msg, err := ws.ReadMessage()
 		s.Type = msgType
 		if err != nil {
 			//客户端断开返回错误，断开当前连接
 			//g.Log().Error(ctx, err)
 			break
 		}
-		s.OnMessage(m.Get(id).(*WebsocketData), msg, msgType)
+		s.OnMessage(conn, msg, msgType)
 	}
 	//关闭连接触发
-	s.OnClose(data)
+
+	s.OnClose(conn)
 	g.Log().Debugf(ctx, "断开连接:uuid=%v,ip=%v", id, ip)
 }
 
@@ -205,17 +209,18 @@ func (s *SocketV1) Uid2Uuid(uid int64) (uuid string) {
 //	@receiver s
 //	@param uid
 //	@param data
-func (s *SocketV1) SendUuid(cmd int32, id uuid.UUID, req proto.Message) {
-	if !m.Contains(id) {
+func (s *SocketV1) SendUuid(cmd int32, uuidStr string, req proto.Message) {
+	if !m.Contains(uuidStr) {
 		return
 	}
+	//格式化数据
 	var data, err = proto.Marshal(req)
 	if err != nil {
 		g.Log().Error(gctx.New(), err)
 		return
 	}
 
-	conn := m.Get(id).(*WebsocketData)
+	conn := m.Get(uuidStr).(*WebsocketData)
 
 	//前置方法
 
@@ -223,9 +228,7 @@ func (s *SocketV1) SendUuid(cmd int32, id uuid.UUID, req proto.Message) {
 		temp := v(cmd, data, 0, "")
 		data, _ = proto.Marshal(temp)
 	}
-
 	conn.Ws.WriteMessage(s.Type, data)
-
 	return
 }
 
@@ -242,26 +245,7 @@ func (s *SocketV1) Send(cmd int32, uid int64, req proto.Message) {
 	if uuid == "" {
 		return
 	}
-	if !m.Contains(uuid) {
-		return
-	}
-
-	//格式化数据
-	var data, err = proto.Marshal(req)
-	if err != nil {
-		g.Log().Error(gctx.New(), err)
-		return
-	}
-
-	conn := m.Get(uuid).(*WebsocketData)
-
-	//前置方法
-
-	for _, v := range Byte2Pb {
-		temp := v(cmd, data, 0, "")
-		data, _ = proto.Marshal(temp)
-	}
-	conn.Ws.WriteMessage(s.Type, data)
+	s.SendUuid(cmd, uuid, req)
 	return
 }
 
@@ -287,6 +271,47 @@ func (s *SocketV1) SendAll(cmd int32, req proto.Message) {
 	})
 }
 
+//加入群组
+func (s *SocketV1) JoinGroup(conn *WebsocketData, group string) {
+	conn.Groups = append(conn.Groups, group)
+	cacheKey := "websocket:group:" + group
+	get, _ := aycache.New("redis").Get(conn.Ctx, cacheKey)
+	var list = make(map[int64]string)
+	if !get.IsNil() {
+		get.Scan(&list)
+	}
+	list[conn.Uid] = conn.Uuid
+	cache.Set(conn.Ctx, cacheKey, list, time.Hour*24*7)
+}
+
+// 退出群组
+func (s *SocketV1) LeaveGroup(conn *WebsocketData, group string) {
+	conn.Groups = tools.RemoveSlice[string](conn.Groups, group)
+	cacheKey := "websocket:group:" + group
+	get, _ := cache.Get(conn.Ctx, cacheKey)
+	var list = make(map[int64]string)
+	if !get.IsNil() {
+		get.Scan(&list)
+	}
+	delete(list, conn.Uid)
+	cache.Set(conn.Ctx, cacheKey, list, time.Hour*24*7)
+}
+
+//群组广播
+func (s *SocketV1) SendGroup(cmd int32, group string, req proto.Message) {
+	cacheKey := "websocket:group:" + group
+	get, _ := cache.Get(gctx.New(), cacheKey)
+	var list = make(map[int64]string)
+	if !get.IsNil() {
+		get.Scan(&list)
+	}
+	for uid, v := range list {
+		if m.Contains(v) {
+			s.Send(cmd, uid, req)
+		}
+	}
+}
+
 // OnClose
 //
 //	@Description:
@@ -303,6 +328,9 @@ func (s *SocketV1) OnClose(conn *WebsocketData) {
 	uid := conn.Uid
 	if uid > 0 {
 		s.UnBindUid(uid)
+		for _, v := range conn.Groups {
+			s.LeaveGroup(conn, v)
+		}
 	}
 
 	// 可能的后续操作：

tools/tools.go

@@ -128,7 +128,7 @@ func (m *tools) Items2Map(items [][]int64) (list map[int64]int64) {
 // 该函数通过遍历切片，从后向前检查每个元素，如果找到与指定值相等的元素，则将其从切片中移除。
 // 这种从后向前的遍历方法可以避免因移除元素而导致的数组重新排列带来的额外计算。
 // RemoveSlice 删除切片中的某个值
-func RemoveSlice[t Number](slice []t, value ...t) []t {
+func RemoveSlice[t Any](slice []t, value ...t) []t {
 	// 从后向前遍历切片
 	for i := len(slice) - 1; i >= 0; i-- {
 		// 检查当前元素是否等于需要移除的值
@@ -149,7 +149,7 @@ func RemoveSlice[t Number](slice []t, value ...t) []t {
 //	@param value 需要查找的值
 //	@param array 进行查找的切片
 //	@return bool 返回是否存在
-func InArray[t Number](array []t, value t) bool {
+func InArray[t Any](array []t, value t) bool {
 	for _, v := range array {
 		if v == value {
 			return true