Compare commits

...

1 Commits

Author SHA1 Message Date
ayflying
366ddb45ea 修改导表工具,修改websocket协议的钩子 2025-04-09 16:50:54 +08:00
3 changed files with 54 additions and 24 deletions

View File

@@ -111,8 +111,12 @@ func (s *Excel) RemoveComments(list []interface{}, json []string) []interface{}
// 遍历当前元素的每个键值对
for _, v3 := range v2.(g.Map) {
// 如果字符串中存在//则跳过不写入temp
if gstr.Contains(gconv.String(v3), "//") {
//delKey = append(delKey, k2)
//if gstr.Contains(gconv.String(v3), "//") {
// //delKey = append(delKey, k2)
// add = false
// break
//}
if strings.HasPrefix(gconv.String(v3), "//") {
add = false
break
}

View File

@@ -1,16 +1,26 @@
package websocket
import "context"
// 定义一个处理方法的类型
type Handler func(ctx context.Context, req any) (err error)
type Handler func(conn *WebsocketData, req any) (err error)
type OnConnectHandler func(conn *WebsocketData)
// 路由器的处理映射
var (
handlers = make(map[int]Handler)
OnConnectHandlers = make([]OnConnectHandler, 0)
OnCloseHandlers = make([]OnConnectHandler, 0)
)
// 注册方法将某个消息路由器ID和对应的处理方法关联起来
func (s *SocketV1) RegisterRouter(cmd int, handler Handler) {
handlers[cmd] = handler
}
//注册方法,讲长连接登陆方法进行注册
func (s *SocketV1) RegisterOnConnect(_func OnConnectHandler) {
OnConnectHandlers = append(OnConnectHandlers, _func)
}
func (s *SocketV1) RegisterOnClose(_func OnConnectHandler) {
OnCloseHandlers = append(OnCloseHandlers, _func)
}

View File

@@ -4,16 +4,18 @@ import (
"context"
"github.com/gogf/gf/v2/container/gmap"
"github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/os/gctx"
"github.com/gogf/gf/v2/net/ghttp"
"github.com/gogf/gf/v2/os/glog"
"github.com/gogf/gf/v2/util/gconv"
"github.com/google/uuid"
"github.com/gorilla/websocket"
"strconv"
"sync"
"github.com/gogf/gf/v2/net/ghttp"
)
type SocketV1 struct{}
type SocketV1 struct {
Type int `json:"type"`
}
var (
//ctx = gctx.New()
@@ -31,7 +33,9 @@ type WebsocketData struct {
}
func NewV1() *SocketV1 {
return &SocketV1{}
return &SocketV1{
Type: 1,
}
}
type SocketInterface interface {
@@ -79,7 +83,6 @@ func (s *SocketV1) OnConnect(ctx context.Context, conn *websocket.Conn) {
defer conn.Close()
id, _ := uuid.NewUUID()
ip := conn.RemoteAddr().String()
data := &WebsocketData{
Uuid: id,
Ws: conn,
@@ -92,6 +95,11 @@ func (s *SocketV1) OnConnect(ctx context.Context, conn *websocket.Conn) {
//to := fmt.Sprintf("创建连接:%v,ip=%v", id, ip)
//s.Send(id, []byte(to))
//用户登录钩子执行
for _, connect := range OnConnectHandlers {
connect(data)
}
for {
//进入当前连接线程拥堵
msgType, msg, err := conn.ReadMessage()
@@ -102,7 +110,7 @@ func (s *SocketV1) OnConnect(ctx context.Context, conn *websocket.Conn) {
s.OnMessage(m.Get(id).(*WebsocketData), msg, msgType)
}
//关闭连接触发
s.OnClose(id, conn)
s.OnClose(data)
g.Log().Debugf(ctx, "断开连接:uuid=%v,ip=%v", id, ip)
}
@@ -113,21 +121,23 @@ func (s *SocketV1) OnConnect(ctx context.Context, conn *websocket.Conn) {
// @param msg
// @param msgType
func (s *SocketV1) OnMessage(conn *WebsocketData, req []byte, msgType int) {
s.Type = msgType
//g.Log().Debugf(ctx, "收到消息:%v,type=%v,conn=%v", string(req), msgType, conn)
//s.Send(conn.Uuid, msg)
//s.SendAll(msg)
msgStr := string(req)
msg := msgStr[8:]
cmd := gconv.Int(msgStr[:8])
//msgStr := string(req)
msg := req[8:]
cmd := gconv.Int(req[:8])
//GetRouter(cmd, conn.Uid, msg)
handler, exist := handlers[cmd]
if exist {
//匹配上路由器
handler(conn.Ctx, msg)
err := handler(conn, msg)
g.Log().Error(conn.Ctx, err)
} else {
//fmt.Println("未注册的路由器ID:", cmd)
s.Send(conn.Uuid, []byte("未注册的协议号:"+msgStr[:8]))
s.OnClose(conn.Uuid, conn.Ws)
s.Send(conn.Uuid, []byte("未注册的协议号:"+strconv.Itoa(cmd)))
s.OnClose(conn)
return
}
@@ -146,7 +156,7 @@ func (s *SocketV1) Send(id uuid.UUID, data []byte) (err error) {
}
conn := m.Get(id).(*WebsocketData)
conn.Ws.WriteMessage(1, data)
conn.Ws.WriteMessage(s.Type, data)
return
}
@@ -156,7 +166,8 @@ func (s *SocketV1) SendAll(data []byte) {
m.Iterator(func(k interface{}, v interface{}) bool {
//fmt.Printf("%v:%v ", k, v)
conn := v.(*WebsocketData)
conn.Ws.WriteMessage(1, data)
conn.Ws.WriteMessage(s.Type, data)
return true
})
}
@@ -166,14 +177,19 @@ func (s *SocketV1) SendAll(data []byte) {
// @Description:
// @receiver s
// @param conn
func (s *SocketV1) OnClose(id uuid.UUID, conn *websocket.Conn) {
func (s *SocketV1) OnClose(conn *WebsocketData) {
// 在此处编写断开连接后的处理逻辑
g.Log().Debugf(gctx.New(), "WebSocket connection from %s has been closed.", conn.RemoteAddr())
//g.Log().Debugf(gctx.New(), "WebSocket connection from %s has been closed.", conn.RemoteAddr())
//用户登录钩子执行
for _, connect := range OnCloseHandlers {
connect(conn)
}
// 可能的后续操作:
// 1. 更新连接状态或从连接池移除
// 2. 发送通知或清理关联资源
// 3. 执行特定于业务的断开处理
m.Remove(id)
conn.Close()
m.Remove(conn.Uuid)
conn.Ws.Close()
}