bl/common/socket/ServerEvent.go

package socket

import (
	"blazing/common/socket/codec"
	"blazing/cool"
	"blazing/logic/service/player"
	"blazing/modules/config/service"
	"bytes"
	"context"
	"encoding/binary"
	"errors"
	"log"
	"os"
	"sync/atomic"
	"time"

	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/os/gtime"
	"github.com/panjf2000/gnet/v2"
)

const (
	minPacketLen = 17
	maxPacketLen = 10 * 1024
)

func (s *Server) Boot(serverid, port uint32) error {
	//	go s.bootws()
	s.serverid = serverid
	s.port = port

	err := gnet.Run(s, s.network+"://"+s.addr,
		gnet.WithMulticore(true),
		gnet.WithTicker(true),

		// 其他调优配置↓
		gnet.WithTCPNoDelay(gnet.TCPNoDelay), // 禁用Nagle算法（降低延迟，适合小数据包场景）
		//gnet.WithReusePort(true),             // 开启SO_REUSEPORT（多核下提升并发）
		//gnet.WithReadBufferCap(1024*64),      // 读缓冲区64KB（根据业务调整，默认太小）
		//	gnet.WithWriteBufferCap(1024*64),     // 写缓冲区64KB

		//gnet.WithLockOSThread(true), // 绑定goroutine到OS线程（减少上下文切换）
	)
	if err != nil {
		panic(err)
	}

	return nil
}

func (s *Server) Stop() error {
	_ = s.eng.Stop(context.Background())
	s.workerPool.Release()

	return nil
}

func (s *Server) OnClose(c gnet.Conn, err error) (action gnet.Action) {
	defer func() {
		if err := recover(); err != nil { // 恢复 panic，err 为 panic 错误值
			if t, ok := c.Context().(*player.ClientData); ok {
				if t.Player != nil {
					if t.Player.Info != nil {
						cool.Logger.Error(context.TODO(), "OnClose 错误：", cool.Config.ServerInfo.OnlineID, t.Player.Info.UserID, err)
						go t.Player.SaveOnDisconnect()
					}

				}
			} else {
				cool.Logger.Error(context.TODO(), "OnClose 错误：", cool.Config.ServerInfo.OnlineID, err)
			}
		}
	}()

	atomic.AddInt64(&cool.Connected, -1)

	v, _ := c.Context().(*player.ClientData)
	if v != nil {
		v.Close()
		if v.Player != nil {
			v.Player.Save() //保存玩家数据
		}
	}
	return
}

func (s *Server) OnTick() (delay time.Duration, action gnet.Action) {
	g.Log().Async().Info(context.Background(), gtime.Now().ISO8601(), "服务器ID", cool.Config.ServerInfo.OnlineID, "链接数", atomic.LoadInt64(&cool.Connected))
	if s.quit && atomic.LoadInt64(&cool.Connected) == 0 {
		os.Exit(0)
	}
	return 30 * time.Second, gnet.None
}

func (s *Server) OnBoot(eng gnet.Engine) gnet.Action {
	s.eng = eng
	service.NewServerService().SetServerID(s.serverid, s.port)
	return gnet.None
}

func (s *Server) OnOpen(conn gnet.Conn) (out []byte, action gnet.Action) {
	if s.network != "tcp" {
		return nil, gnet.Close
	}
	if conn.Context() == nil {
		conn.SetContext(player.NewClientData(conn))
	}
	atomic.AddInt64(&cool.Connected, 1)
	return nil, gnet.None
}

func (s *Server) OnTraffic(c gnet.Conn) (action gnet.Action) {
	defer func() {
		if err := recover(); err != nil {
			if t, ok := c.Context().(*player.ClientData); ok {
				if t.Player != nil && t.Player.Info != nil {
					cool.Logger.Error(context.TODO(), "OnTraffic 错误：", cool.Config.ServerInfo.OnlineID, t.Player.Info.UserID, err)
					t.Player.Service.Info.Save(*t.Player.Info)
				}
			}
		}
	}()

	client := c.Context().(*player.ClientData)
	if s.discorse && !client.IsCrossDomainChecked() {
		handled, ready, action := handle(c)
		if action != gnet.None {
			return action
		}
		if handled {
			client.MarkCrossDomainChecked()
			return gnet.None
		}
		if !ready {
			return gnet.None
		}
		client.MarkCrossDomainChecked()
	}

	ws := client.Wsmsg
	if ws.Tcp {
		return s.handleTCP(c)
	}

	readAction, inboundLen := ws.ReadBufferBytes(c)
	if readAction == gnet.Close {
		return gnet.Close
	}

	state, action := ws.Upgrade(c)
	if action != gnet.None {
		return action
	}
	if state == player.UpgradeNeedMoreData {
		return gnet.None
	}
	if state == player.UpgradeUseTCP {
		return s.handleTCP(c)
	}

	if inboundLen > 0 {
		if _, err := c.Discard(inboundLen); err != nil {
			return gnet.Close
		}
		ws.ResetInboundMirror()
	}

	messages, err := ws.Decode(c)
	if err != nil {
		return gnet.Close
	}
	if messages == nil {
		return
	}

	for _, msg := range messages {
		if !s.onevent(c, msg.Payload) {
			return gnet.Close
		}
	}
	return gnet.None
}

func (s *Server) handleTCP(conn gnet.Conn) (action gnet.Action) {
	client := conn.Context().(*player.ClientData)
	if s.discorse && !client.IsCrossDomainChecked() {
		handled, ready, action := handle(conn)
		if action != gnet.None {
			return action
		}
		if !ready {
			return gnet.None
		}
		if handled {
			client.MarkCrossDomainChecked()
			return gnet.None
		}
		client.MarkCrossDomainChecked()
	}

	body, err := s.codec.Decode(conn)
	if err != nil {
		if errors.Is(err, codec.ErrIncompletePacket) {
			return gnet.None
		}
		return gnet.Close
	}
	if !s.onevent(conn, body) {
		return gnet.Close
	}
	if conn.InboundBuffered() > 0 {
		if err := conn.Wake(nil); err != nil {
			return gnet.Close
		}
	}
	return action
}

const CROSS_DOMAIN = "<?xml version=\"1.0\"?><!DOCTYPE cross-domain-policy><cross-domain-policy><allow-access-from domain=\"*\" to-ports=\"*\" /></cross-domain-policy>\x00"
const TEXT = "<policy-file-request/>\x00"

func handle(c gnet.Conn) (handled bool, ready bool, action gnet.Action) {
	probeLen := c.InboundBuffered()
	if probeLen == 0 {
		return false, false, gnet.None
	}
	if probeLen > len(TEXT) {
		probeLen = len(TEXT)
	}

	data, err := c.Peek(probeLen)
	if err != nil {
		log.Printf("Error reading cross-domain request: %v", err)
		return false, false, gnet.Close
	}
	if !bytes.Equal(data, []byte(TEXT[:probeLen])) {
		return false, true, gnet.None
	}
	if probeLen < len(TEXT) {
		return false, false, gnet.None
	}
	if _, err := c.Write([]byte(CROSS_DOMAIN)); err != nil {
		return false, true, gnet.Close
	}
	if _, err := c.Discard(len(TEXT)); err != nil {
		return false, true, gnet.Close
	}
	return true, true, gnet.None
}

func (s *Server) onevent(c gnet.Conn, v []byte) bool {
	if !isValidPacket(v) {
		return false
	}
	if t, ok := c.Context().(*player.ClientData); ok {
		t.PushEvent(v, s.workerPool.Submit)
	}
	return true
}

func isValidPacket(v []byte) bool {
	if len(v) < minPacketLen || len(v) > maxPacketLen {
		return false
	}
	return binary.BigEndian.Uint32(v[0:4]) == uint32(len(v))
}