admin/bl
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(socket): 重构 socket 模块

Browse Source 
- 移除 common/data/socket 目录下的大部分文件
- 新增 service 目录，将 Player 和 Conn 结构体移至该目录
- 更新 LogicClient 中的方法签名，使用 service 包的类型
- 重构 Controller 中的方法，适应新的 service 包结构
This commit is contained in:
昔念
2025-09-04 02:00:57 +08:00
parent afdf015d62
commit 621a9d3858
50 changed files with 886 additions and 849 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
common/data/socket/Client.go
View File

@@ -1,30 +0,0 @@
package socket
func ConutPlayer() int {
count := 0
Mainplayer.Range(func(uint32, *Player) bool {
count++
return true // 继续遍历
})
return count
}
type ClientData struct {
IsCrossDomain bool //是否跨域过
Player *Player //客户实体
//UserID uint32
Wsmsg *WsCodec
}
func NewClientData() *ClientData {
cd := ClientData{
IsCrossDomain: false,
Player: nil,
Wsmsg: &WsCodec{},
}
return &cd
}

210
common/data/socket/player.go
View File

@@ -1,210 +0,0 @@
package socket
import (
"blazing/common/utils"
"blazing/modules/blazing/model"
"blazing/modules/blazing/service"
"context"
"fmt"
"sync"
"time"
"github.com/gobwas/ws"
"github.com/gobwas/ws/wsutil"
"github.com/gogf/gf/v2/frame/g"
"github.com/panjf2000/gnet/pkg/logging"
"github.com/tnnmigga/enum"
)
var Mainplayer = &utils.SyncMap[uint32, *Player]{} //玩家数据
func (c *Conn) SendPack(bytes []byte) error {
if t, ok := c.MainConn.Context().(*ClientData); ok {
if t.Wsmsg.Upgraded {
// This is the echo server
err := wsutil.WriteServerMessage(c.MainConn, ws.OpBinary, bytes)
if err != nil {
logging.Infof("conn[%v] [err=%v]", c.MainConn.RemoteAddr().String(), err.Error())
return err
}
} else {
_, err := c.MainConn.Write(bytes)
if err != nil {
logging.Error(err)
}
}
}
return nil
}
type OgreInfo struct {
Data [9]OgrePetInfo
}
type OgrePetInfo struct {
Id uint32
Shiny uint32
Lv uint32 `struc:"skip"` //等级
}
type Player struct {
MainConn *Conn
IsLogin bool //是否登录
mu sync.Mutex
loginChan chan struct{} // 登录完成通知通道
Info *model.PlayerInfo
StopChan chan struct{} //停止刷怪协程
context.Context
Playerinvite uint32 //当前邀请的玩家ID
Onlinetime uint32 //当前登录时间
OgreInfo *OgreInfo
FightID string //绑定战斗标识 替代本身的是否战斗标记 //IsFighting bool
//FightInfo info.NoteReadyToFightInfo
}
// PlayerOption 定义配置 Player 的函数类型
type PlayerOption func(*Player)
func WithConn(c *Conn) PlayerOption {
return func(p *Player) {
p.MainConn = c
}
}
// NewPlayer 使用 Options 模式创建 Player 实例
func NewPlayer(opts ...PlayerOption) *Player {
p := &Player{
loginChan: make(chan struct{}),
}
for _, opt := range opts {
opt(p)
}
return p
}
func (p *Player) SendPack(b []byte) error {
// fmt.Println("发送数据包", len(b))
err := p.MainConn.SendPack(b)
return err
}
func (p *Player) Cheak(b error) {
if b != nil {
g.Log().Error(context.Background(), "出现错误", p.Info.UserID, b.Error())
}
}
// Save 保存玩家数据
func (p *Player) Save() {
p.Info.TimeToday = p.Info.TimeToday + uint32(time.Now().Unix()) - uint32(p.Onlinetime) //保存电池时间
p.Onlinetime = uint32(time.Now().Unix())
service.NewUserService(p.Info.UserID).Save(p.Info)
}
// IsLoggedIn 检查是否已登录
func (lw *Player) IsLoggedIn() bool {
lw.mu.Lock()
defer lw.mu.Unlock()
return lw.IsLogin
}
// WaitForLogin 等待登录完成，无超时
func (lw *Player) WaitForLogin() error {
if lw.IsLoggedIn() {
return nil
}
// 阻塞等待登录完成
<-lw.loginChan
return nil
}
// WaitForLoginWithTimeout 带超时的登录等待
func (lw *Player) WaitForLoginWithTimeout(timeout time.Duration) error {
if lw.IsLoggedIn() {
return nil
}
// 使用定时器实现超时
timer := time.NewTimer(timeout)
defer timer.Stop()
select {
case <-lw.loginChan:
return nil
case <-timer.C:
return fmt.Errorf("登录等待超时: %v", timeout)
}
}
// WaitForLoginWithCtx 带上下文的登录等待
func (lw *Player) WaitForLoginWithCtx(ctx context.Context) error {
if lw.IsLoggedIn() {
return nil
}
select {
case <-lw.loginChan:
return nil
case <-ctx.Done():
return ctx.Err() // 上下文取消或超时
}
}
// CompleteLogin 标记登录完成并通知等待者
func (lw *Player) CompleteLogin() {
lw.mu.Lock()
defer lw.mu.Unlock()
if !lw.IsLogin {
lw.IsLogin = true
close(lw.loginChan) // 关闭通道以通知所有等待者
}
}
// 战斗模式
type EnumBattleMode int
var BattleMode_PVP = enum.New[struct {
PVP_1V1 EnumBattleMode `enum:"1"`
PVP_6V6 EnumBattleMode `enum:"2"`
}]()
var Playerinvitemap map[uint32][]Playerinvite = make(map[uint32][]Playerinvite) //玩家邀请信息 ,比如一个玩家被多人邀请对战
type Playerinvite struct { //挂载到[]Playerinvite上? 被邀请者->邀请者
InviteID uint32 // 邀请者
InviteTime EnumBattleMode //游戏模式
}
// 邀请玩家加入战斗 邀请者,被邀请者,邀请模式
func (lw *Player) InvitePlayerToBattle(target int64, mode EnumBattleMode) {
t, ok := Playerinvitemap[uint32(target)] //被邀请者是否被邀请过
if ok { //说明存在被邀请
t = append(t, Playerinvite{uint32(lw.Info.UserID), mode})
Playerinvitemap[uint32(target)] = t
} else {
Playerinvitemap[uint32(target)] = []Playerinvite{{uint32(lw.Info.UserID), mode}}
}
lw.Playerinvite = uint32(target)
}
// 取消对战邀请
func (lw *Player) CancelBattle() {
if lw.Playerinvite == 0 {
return
}
delete(Playerinvitemap, uint32(lw.Playerinvite)) //删除玩家邀请信息
lw.Playerinvite = 0
}

170
common/data/socket/wscodec.go
View File

@@ -1,170 +0,0 @@
package socket
import (
"bytes"
"errors"
"io"
"sync"
"github.com/gobwas/ws"
"github.com/gobwas/ws/wsutil"
"github.com/panjf2000/gnet/v2"
"github.com/panjf2000/gnet/v2/pkg/logging"
)
type Conn struct {
MainConn gnet.Conn `struc:"[0]pad"` //TODO 不序列化,,序列化下面的作为blob存数据库
Mu sync.Mutex
}
func NewConn(c gnet.Conn) *Conn {
return &Conn{MainConn: c}
}
type WsCodec struct {
Upgraded bool // 链接是否升级
Buf bytes.Buffer // 从实际socket中读取到的数据缓存
wsMsgBuf wsMessageBuf // ws 消息缓存
//Isinitws bool
}
type wsMessageBuf struct {
curHeader *ws.Header
cachedBuf bytes.Buffer
}
type readWrite struct {
io.Reader
io.Writer
}
func (w *WsCodec) Upgrade(c gnet.Conn) (ok bool, action gnet.Action) {
if w.Upgraded {
ok = true
return
}
buf := &w.Buf
tmpReader := bytes.NewReader(buf.Bytes())
oldLen := tmpReader.Len()
logging.Infof("do Upgrade")
hs, err := ws.Upgrade(readWrite{tmpReader, c})
skipN := oldLen - tmpReader.Len()
if err != nil {
if err == io.EOF || errors.Is(err, io.ErrUnexpectedEOF) { //数据不完整，不跳过 buf 中的 skipN 字节（此时 buf 中存放的仅是部分 "handshake data" bytes），下次再尝试读取
return
}
buf.Next(skipN)
logging.Errorf("conn[%v] [err=%v]", c.RemoteAddr().String(), err.Error())
action = gnet.Close
return
}
buf.Next(skipN)
logging.Infof("conn[%v] upgrade websocket protocol! Handshake: %v", c.RemoteAddr().String(), hs)
ok = true
w.Upgraded = true
return
}
func (w *WsCodec) ReadBufferBytes(c gnet.Conn) (gnet.Action, int) {
size := c.InboundBuffered()
//buf := make([]byte, size)
read, err := c.Peek(size)
if err != nil {
logging.Errorf("read err! %v", err)
return gnet.Close, 0
}
// if read < size {
// logging.Errorf("read bytes len err! size: %d read: %d", size, read)
// return gnet.Close
// }
w.Buf.Write(read)
return gnet.None, size
}
func (w *WsCodec) Decode(c gnet.Conn) (outs []wsutil.Message, err error) {
// fmt.Println("do Decode")
messages, err := w.readWsMessages()
if err != nil {
logging.Errorf("Error reading message! %v", err)
return nil, err
}
if messages == nil || len(messages) <= 0 { //没有读到完整数据 不处理
return
}
for _, message := range messages {
if message.OpCode.IsControl() {
err = wsutil.HandleClientControlMessage(c, message)
if err != nil {
return
}
continue
}
if message.OpCode == ws.OpText || message.OpCode == ws.OpBinary {
outs = append(outs, message)
}
}
return
}
func (w *WsCodec) readWsMessages() (messages []wsutil.Message, err error) {
msgBuf := &w.wsMsgBuf
in := &w.Buf
for {
// 从 in 中读出 header，并将 header bytes 写入 msgBuf.cachedBuf
if msgBuf.curHeader == nil {
if in.Len() < ws.MinHeaderSize { //头长度至少是2
return
}
var head ws.Header
if in.Len() >= ws.MaxHeaderSize {
head, err = ws.ReadHeader(in)
if err != nil {
return messages, err
}
} else { //有可能不完整，构建新的 reader 读取 head，读取成功才实际对 in 进行读操作
tmpReader := bytes.NewReader(in.Bytes())
oldLen := tmpReader.Len()
head, err = ws.ReadHeader(tmpReader)
skipN := oldLen - tmpReader.Len()
if err != nil {
if err == io.EOF || errors.Is(err, io.ErrUnexpectedEOF) { //数据不完整
return messages, nil
}
in.Next(skipN)
return nil, err
}
in.Next(skipN)
}
msgBuf.curHeader = &head
err = ws.WriteHeader(&msgBuf.cachedBuf, head)
if err != nil {
return nil, err
}
}
dataLen := (int)(msgBuf.curHeader.Length)
// 从 in 中读出 data，并将 data bytes 写入 msgBuf.cachedBuf
if dataLen > 0 {
if in.Len() < dataLen { //数据不完整
logging.Infof("incomplete data")
return
}
_, err = io.CopyN(&msgBuf.cachedBuf, in, int64(dataLen))
if err != nil {
return
}
}
if msgBuf.curHeader.Fin { //当前 header 已经是一个完整消息
messages, err = wsutil.ReadClientMessage(&msgBuf.cachedBuf, messages)
if err != nil {
return nil, err
}
msgBuf.cachedBuf.Reset()
} else {
logging.Infof("The data is split into multiple frames")
}
msgBuf.curHeader = nil
}
}