chenyh
/
vRp.CD2g_test
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor: 调整 server 数据包相关处理函数的接收参数不再为 server.Packet，同时移除 server.Packet

This commit is contained in:
kercylan98 2023-08-23 19:29:55 +08:00
parent 4785c60c5e
commit 4850dd4aa3
31 changed files with 614 additions and 590 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
game/builtin/player.go
View File

@ -34,7 +34,7 @@ func (slf *Player[ID]) UseConn(conn *server.Conn) {
}
// Send 向该玩家发送数据
func (slf *Player[ID]) Send(packet server.Packet) {
func (slf *Player[ID]) Send(packet []byte) {
slf.conn.Write(packet)
}

158
server/client/client.go Normal file
View File

@ -0,0 +1,158 @@
package client
import (
"github.com/kercylan98/minotaur/utils/concurrent"
"sync"
"time"
)
// NewClient 创建客户端
func NewClient(core Core) *Client {
client := &Client{
events: new(events),
core: core,
}
return client
}
// Client 客户端
type Client struct {
*events
core Core
mutex sync.Mutex
packetPool *concurrent.Pool[*Packet]
packets []*Packet
accumulate []*Packet
}
func (slf *Client) Run() error {
var wait = new(sync.WaitGroup)
wait.Add(1)
go slf.writeLoop(wait)
wait.Wait()
var runState = make(chan error)
go func() {
defer func() {
if err := recover(); err != nil {
slf.Close(err.(error))
}
}()
slf.core.Run(runState, slf.onReceive)
}()
err := <-runState
if err != nil {
slf.Close()
return err
}
slf.OnConnectionOpenedEvent(slf)
return nil
}
// IsConnected 是否已连接
func (slf *Client) IsConnected() bool {
return slf.packetPool != nil
}
// Close 关闭
func (slf *Client) Close(err ...error) {
slf.core.Close()
if slf.packetPool != nil {
slf.packetPool.Close()
slf.packetPool = nil
}
slf.packets = nil
if len(err) > 0 {
slf.OnConnectionClosedEvent(slf, err[0])
} else {
slf.OnConnectionClosedEvent(slf, nil)
}
}
// WriteWS 向连接中写入指定 websocket 数据类型
// - wst: websocket模式中指定消息类型
func (slf *Client) WriteWS(wst int, packet []byte, callback ...func(err error)) {
slf.write(wst, packet, callback...)
}
// Write 向连接中写入数据
func (slf *Client) Write(packet []byte, callback ...func(err error)) {
slf.write(0, packet, callback...)
}
// write 向连接中写入数据
// - messageType: websocket模式中指定消息类型
func (slf *Client) write(wst int, packet []byte, callback ...func(err error)) {
cp := slf.packetPool.Get()
cp.wst = wst
cp.data = packet
if len(callback) > 0 {
cp.callback = callback[0]
}
if slf.packetPool == nil {
slf.accumulate = append(slf.accumulate, cp)
return
}
slf.mutex.Lock()
slf.packets = append(slf.packets, cp)
slf.mutex.Unlock()
}
// writeLoop 写循环
func (slf *Client) writeLoop(wait *sync.WaitGroup) {
slf.packetPool = concurrent.NewPool[*Packet](10*1024,
func() *Packet {
return &Packet{}
}, func(data *Packet) {
data.wst = 0
data.data = nil
data.callback = nil
},
)
slf.mutex.Lock()
slf.packets = append(slf.packets, slf.accumulate...)
slf.accumulate = nil
slf.mutex.Unlock()
defer func() {
if err := recover(); err != nil {
slf.Close(err.(error))
}
}()
wait.Done()
for {
slf.mutex.Lock()
if slf.packetPool == nil {
slf.mutex.Unlock()
return
}
if len(slf.packets) == 0 {
slf.mutex.Unlock()
time.Sleep(50 * time.Millisecond)
continue
}
packets := slf.packets[0:]
slf.packets = slf.packets[0:0]
slf.mutex.Unlock()
for i := 0; i < len(packets); i++ {
data := packets[i]
var err = slf.core.Write(data)
callback := data.callback
slf.packetPool.Release(data)
if callback != nil {
callback(err)
}
if err != nil {
panic(err)
}
}
}
}
func (slf *Client) onReceive(wst int, packet []byte) {
slf.OnConnectionReceivePacketEvent(slf, wst, packet)
}
// GetServerAddr 获取服务器地址
func (slf *Client) GetServerAddr() string {
return slf.core.GetServerAddr()
}

17
server/client/client_core.go Normal file
View File

@ -0,0 +1,17 @@
package client
type Core interface {
// Run 启动客户端
// - runState: 运行状态，当客户端启动完成时，应该向该通道发送 error 或 nil
// - receive: 接收到数据包时应该将数据包发送到该函数，wst 表示 websocket 的数据类型，data 表示数据包
Run(runState chan<- error, receive func(wst int, packet []byte))
// Write 向客户端写入数据包
Write(packet *Packet) error
// Close 关闭客户端
Close()
// GetServerAddr 获取服务器地址
GetServerAddr() string
}

46
server/client/client_events.go Normal file
View File

@ -0,0 +1,46 @@
package client
type (
ConnectionClosedEventHandle func(conn *Client, err any)
ConnectionOpenedEventHandle func(conn *Client)
ConnectionReceivePacketEventHandle func(conn *Client, wst int, packet []byte)
)
type events struct {
ConnectionClosedEventHandles []ConnectionClosedEventHandle
ConnectionOpenedEventHandles []ConnectionOpenedEventHandle
ConnectionReceivePacketEventHandles []ConnectionReceivePacketEventHandle
}
// RegConnectionClosedEvent 注册连接关闭事件
func (slf *events) RegConnectionClosedEvent(handle ConnectionClosedEventHandle) {
slf.ConnectionClosedEventHandles = append(slf.ConnectionClosedEventHandles, handle)
}
func (slf *events) OnConnectionClosedEvent(conn *Client, err any) {
for _, handle := range slf.ConnectionClosedEventHandles {
handle(conn, err)
}
}
// RegConnectionOpenedEvent 注册连接打开事件
func (slf *events) RegConnectionOpenedEvent(handle ConnectionOpenedEventHandle) {
slf.ConnectionOpenedEventHandles = append(slf.ConnectionOpenedEventHandles, handle)
}
func (slf *events) OnConnectionOpenedEvent(conn *Client) {
for _, handle := range slf.ConnectionOpenedEventHandles {
handle(conn)
}
}
// RegConnectionReceivePacketEvent 注册连接接收数据包事件
func (slf *events) RegConnectionReceivePacketEvent(handle ConnectionReceivePacketEventHandle) {
slf.ConnectionReceivePacketEventHandles = append(slf.ConnectionReceivePacketEventHandles, handle)
}
func (slf *events) OnConnectionReceivePacketEvent(conn *Client, wst int, packet []byte) {
for _, handle := range slf.ConnectionReceivePacketEventHandles {
handle(conn, wst, packet)
}
}

6
server/client/packet.go
View File

@ -1,7 +1,7 @@
package client
type Packet struct {
websocketMessageType int // websocket 消息类型
packet []byte // 数据包
callback func(err error) // 回调函数
wst int // websocket 的数据类型
data []byte // 数据包
callback func(err error) // 回调函数
}

168
server/client/uds.go
View File

@ -1,156 +1,48 @@
package client
import (
"github.com/kercylan98/minotaur/server"
"github.com/kercylan98/minotaur/utils/concurrent"
"net"
"sync"
"time"
)
// NewUnixDomainSocket 创建 unix domain socket 客户端
func NewUnixDomainSocket(addr string) *UnixDomainSocket {
return &UnixDomainSocket{
udsEvents: new(udsEvents),
addr: addr,
data: map[string]any{},
}
func NewUnixDomainSocket(addr string) *Client {
return NewClient(&UnixDomainSocket{
addr: addr,
})
}
// UnixDomainSocket unix domain socket 客户端
type UnixDomainSocket struct {
*udsEvents
conn net.Conn
addr string
data map[string]any
mutex sync.Mutex
packetPool *concurrent.Pool[*Packet]
packets []*Packet
accumulate []server.Packet
conn net.Conn
addr string
closed bool
}
// Run 启动
func (slf *UnixDomainSocket) Run() error {
func (slf *UnixDomainSocket) Run(runState chan<- error, receive func(wst int, packet []byte)) {
c, err := net.Dial("unix", slf.addr)
if err != nil {
return err
}
slf.conn = c
var wait = new(sync.WaitGroup)
wait.Add(1)
go slf.writeLoop(wait)
wait.Wait()
go func() {
defer func() {
if err := recover(); err != nil {
slf.Close()
slf.OnUDSConnectionClosedEvent(slf, err)
}
}()
slf.OnUDSConnectionOpenedEvent(slf)
packet := make([]byte, 1024)
for slf.packetPool != nil {
n, readErr := slf.conn.Read(packet)
if readErr != nil {
panic(readErr)
}
slf.OnUDSConnectionReceivePacketEvent(slf, server.NewPacket(packet[:n]))
}
}()
return nil
}
// Close 关闭
func (slf *UnixDomainSocket) Close() {
if slf.packetPool != nil {
slf.packetPool.Close()
slf.packetPool = nil
}
slf.packets = nil
}
// IsConnected 是否已连接
func (slf *UnixDomainSocket) IsConnected() bool {
return slf.packetPool != nil
}
// GetData 获取数据
func (slf *UnixDomainSocket) GetData(key string) any {
return slf.data[key]
}
// SetData 设置数据
func (slf *UnixDomainSocket) SetData(key string, value any) {
slf.data[key] = value
}
// Write 向连接中写入数据
func (slf *UnixDomainSocket) Write(packet server.Packet) {
if slf.packetPool == nil {
slf.accumulate = append(slf.accumulate, packet)
runState <- err
return
}
cp := slf.packetPool.Get()
cp.websocketMessageType = packet.WebsocketType
cp.packet = packet.Data
slf.mutex.Lock()
slf.packets = append(slf.packets, cp)
slf.mutex.Unlock()
slf.conn = c
runState <- nil
packet := make([]byte, 1024)
for !slf.closed {
n, readErr := slf.conn.Read(packet)
if readErr != nil {
panic(readErr)
}
receive(0, packet[:n])
}
}
// writeLoop 写循环
func (slf *UnixDomainSocket) writeLoop(wait *sync.WaitGroup) {
slf.packetPool = concurrent.NewPool[*Packet](10*1024,
func() *Packet {
return &Packet{}
}, func(data *Packet) {
data.packet = nil
data.websocketMessageType = 0
data.callback = nil
},
)
slf.mutex.Lock()
for _, packet := range slf.accumulate {
cp := slf.packetPool.Get()
cp.websocketMessageType = packet.WebsocketType
cp.packet = packet.Data
slf.packets = append(slf.packets, cp)
}
slf.accumulate = nil
slf.mutex.Unlock()
defer func() {
if err := recover(); err != nil {
slf.Close()
}
}()
wait.Done()
for {
slf.mutex.Lock()
if slf.packetPool == nil {
slf.mutex.Unlock()
return
}
if len(slf.packets) == 0 {
slf.mutex.Unlock()
time.Sleep(50 * time.Millisecond)
continue
}
packets := slf.packets[0:]
slf.packets = slf.packets[0:0]
slf.mutex.Unlock()
for i := 0; i < len(packets); i++ {
data := packets[i]
var _, err = slf.conn.Write(data.packet)
callback := data.callback
slf.packetPool.Release(data)
if callback != nil {
callback(err)
}
if err != nil {
panic(err)
}
}
}
func (slf *UnixDomainSocket) Write(packet *Packet) error {
_, err := slf.conn.Write(packet.data)
return err
}
func (slf *UnixDomainSocket) Close() {
slf.closed = true
}
func (slf *UnixDomainSocket) GetServerAddr() string {
return slf.addr
}

48
server/client/uds_events.go
View File

@ -1,48 +0,0 @@
package client
import "github.com/kercylan98/minotaur/server"
type (
UDSConnectionClosedEventHandle func(conn *UnixDomainSocket, err any)
UDSConnectionOpenedEventHandle func(conn *UnixDomainSocket)
UDSConnectionReceivePacketEventHandle func(conn *UnixDomainSocket, packet server.Packet)
)
type udsEvents struct {
UDSConnectionClosedEventHandles []UDSConnectionClosedEventHandle
UDSConnectionOpenedEventHandles []UDSConnectionOpenedEventHandle
UDSConnectionReceivePacketEventHandles []UDSConnectionReceivePacketEventHandle
}
// RegUDSConnectionClosedEvent 注册连接关闭事件
func (slf *udsEvents) RegUDSConnectionClosedEvent(handle UDSConnectionClosedEventHandle) {
slf.UDSConnectionClosedEventHandles = append(slf.UDSConnectionClosedEventHandles, handle)
}
func (slf *udsEvents) OnUDSConnectionClosedEvent(conn *UnixDomainSocket, err any) {
for _, handle := range slf.UDSConnectionClosedEventHandles {
handle(conn, err)
}
}
// RegUDSConnectionOpenedEvent 注册连接打开事件
func (slf *udsEvents) RegUDSConnectionOpenedEvent(handle UDSConnectionOpenedEventHandle) {
slf.UDSConnectionOpenedEventHandles = append(slf.UDSConnectionOpenedEventHandles, handle)
}
func (slf *udsEvents) OnUDSConnectionOpenedEvent(conn *UnixDomainSocket) {
for _, handle := range slf.UDSConnectionOpenedEventHandles {
handle(conn)
}
}
// RegUDSConnectionReceivePacketEvent 注册连接接收数据包事件
func (slf *udsEvents) RegUDSConnectionReceivePacketEvent(handle UDSConnectionReceivePacketEventHandle) {
slf.UDSConnectionReceivePacketEventHandles = append(slf.UDSConnectionReceivePacketEventHandles, handle)
}
func (slf *udsEvents) OnUDSConnectionReceivePacketEvent(conn *UnixDomainSocket, packet server.Packet) {
for _, handle := range slf.UDSConnectionReceivePacketEventHandles {
handle(conn, packet)
}
}

16
server/client/uds_test.go
View File

@ -8,21 +8,21 @@ import (
)
func TestUnixDomainSocket_Write(t *testing.T) {
var close = make(chan struct{})
var closed = make(chan struct{})
srv := server.New(server.NetworkUnix)
srv.RegConnectionReceivePacketEvent(func(srv *server.Server, conn *server.Conn, packet server.Packet) {
t.Log(packet)
srv.RegConnectionReceivePacketEvent(func(srv *server.Server, conn *server.Conn, packet []byte) {
t.Log(string(packet))
conn.Write(packet)
})
srv.RegStartFinishEvent(func(srv *server.Server) {
time.Sleep(time.Second)
cli := client.NewUnixDomainSocket("./test.sock")
cli.RegUDSConnectionOpenedEvent(func(conn *client.UnixDomainSocket) {
conn.Write(server.NewPacketString("Hello~"))
cli.RegConnectionOpenedEvent(func(conn *client.Client) {
conn.Write([]byte("Hello~"))
})
cli.RegUDSConnectionReceivePacketEvent(func(conn *client.UnixDomainSocket, packet server.Packet) {
cli.RegConnectionReceivePacketEvent(func(conn *client.Client, wst int, packet []byte) {
t.Log(packet)
close <- struct{}{}
closed <- struct{}{}
})
if err := cli.Run(); err != nil {
panic(err)
@ -34,6 +34,6 @@ func TestUnixDomainSocket_Write(t *testing.T) {
}
}()
<-close
<-closed
srv.Shutdown()
}

165
server/client/websocket.go
View File

@ -3,155 +3,50 @@ package client
import (
"github.com/gorilla/websocket"
"github.com/kercylan98/minotaur/server"
"github.com/kercylan98/minotaur/utils/concurrent"
"sync"
"time"
)
// NewWebsocket 创建 websocket 客户端
func NewWebsocket(addr string) *Websocket {
client := &Websocket{
websocketEvents: new(websocketEvents),
addr: addr,
data: map[string]any{},
}
return client
func NewWebsocket(addr string) *Client {
return NewClient(&Websocket{
addr: addr,
})
}
// Websocket websocket 客户端
type Websocket struct {
*websocketEvents
conn *websocket.Conn
addr string
data map[string]any
mutex sync.Mutex
packetPool *concurrent.Pool[*Packet]
packets []*Packet
accumulate []server.Packet
addr string
conn *websocket.Conn
clsoed bool
}
// Run 启动
func (slf *Websocket) Run() error {
func (slf *Websocket) Run(runState chan<- error, receive func(wst int, packet []byte)) {
ws, _, err := websocket.DefaultDialer.Dial(slf.addr, nil)
if err != nil {
return err
}
slf.conn = ws
var wait = new(sync.WaitGroup)
wait.Add(1)
go slf.writeLoop(wait)
wait.Wait()
go func() {
defer func() {
if err := recover(); err != nil {
slf.Close()
slf.OnWebsocketConnectionClosedEvent(slf, err)
}
}()
slf.OnWebsocketConnectionOpenedEvent(slf)
for slf.packetPool != nil {
messageType, packet, readErr := ws.ReadMessage()
if readErr != nil {
panic(readErr)
}
slf.OnWebsocketConnectionReceivePacketEvent(slf, server.NewWSPacket(messageType, packet))
}
}()
return nil
}
// Close 关闭
func (slf *Websocket) Close() {
if slf.packetPool != nil {
slf.packetPool.Close()
slf.packetPool = nil
}
slf.packets = nil
}
// IsConnected 是否已连接
func (slf *Websocket) IsConnected() bool {
return slf.packetPool != nil
}
// GetData 获取数据
func (slf *Websocket) GetData(key string) any {
return slf.data[key]
}
// SetData 设置数据
func (slf *Websocket) SetData(key string, value any) {
slf.data[key] = value
}
// Write 向连接中写入数据
// - messageType: websocket模式中指定消息类型
func (slf *Websocket) Write(packet server.Packet) {
if slf.packetPool == nil {
slf.accumulate = append(slf.accumulate, packet)
runState <- err
return
}
cp := slf.packetPool.Get()
cp.websocketMessageType = packet.WebsocketType
cp.packet = packet.Data
slf.mutex.Lock()
slf.packets = append(slf.packets, cp)
slf.mutex.Unlock()
slf.conn = ws
runState <- nil
for !slf.clsoed {
messageType, packet, readErr := ws.ReadMessage()
if readErr != nil {
panic(readErr)
}
receive(messageType, packet)
}
}
// writeLoop 写循环
func (slf *Websocket) writeLoop(wait *sync.WaitGroup) {
slf.packetPool = concurrent.NewPool[*Packet](10*1024,
func() *Packet {
return &Packet{}
}, func(data *Packet) {
data.packet = nil
data.websocketMessageType = 0
data.callback = nil
},
)
slf.mutex.Lock()
for _, packet := range slf.accumulate {
cp := slf.packetPool.Get()
cp.websocketMessageType = packet.WebsocketType
cp.packet = packet.Data
slf.packets = append(slf.packets, cp)
}
slf.accumulate = nil
slf.mutex.Unlock()
defer func() {
if err := recover(); err != nil {
slf.Close()
}
}()
wait.Done()
for {
slf.mutex.Lock()
if slf.packetPool == nil {
slf.mutex.Unlock()
return
}
if len(slf.packets) == 0 {
slf.mutex.Unlock()
time.Sleep(50 * time.Millisecond)
continue
}
packets := slf.packets[0:]
slf.packets = slf.packets[0:0]
slf.mutex.Unlock()
for i := 0; i < len(packets); i++ {
data := packets[i]
var err = slf.conn.WriteMessage(data.websocketMessageType, data.packet)
callback := data.callback
slf.packetPool.Release(data)
if callback != nil {
callback(err)
}
if err != nil {
panic(err)
}
}
func (slf *Websocket) Write(packet *Packet) error {
if packet.wst == 0 {
packet.wst = server.WebsocketMessageTypeBinary
}
return slf.conn.WriteMessage(packet.wst, packet.data)
}
func (slf *Websocket) Close() {
slf.clsoed = true
}
func (slf *Websocket) GetServerAddr() string {
return slf.addr
}

48
server/client/websocket_events.go
View File

@ -1,48 +0,0 @@
package client
import "github.com/kercylan98/minotaur/server"
type (
WebsocketConnectionClosedEventHandle func(conn *Websocket, err any)
WebsocketConnectionOpenedEventHandle func(conn *Websocket)
WebsocketConnectionReceivePacketEventHandle func(conn *Websocket, packet server.Packet)
)
type websocketEvents struct {
websocketConnectionClosedEventHandles []WebsocketConnectionClosedEventHandle
websocketConnectionOpenedEventHandles []WebsocketConnectionOpenedEventHandle
websocketConnectionReceivePacketEventHandles []WebsocketConnectionReceivePacketEventHandle
}
// RegWebsocketConnectionClosedEvent 注册连接关闭事件
func (slf *websocketEvents) RegWebsocketConnectionClosedEvent(handle WebsocketConnectionClosedEventHandle) {
slf.websocketConnectionClosedEventHandles = append(slf.websocketConnectionClosedEventHandles, handle)
}
func (slf *websocketEvents) OnWebsocketConnectionClosedEvent(conn *Websocket, err any) {
for _, handle := range slf.websocketConnectionClosedEventHandles {
handle(conn, err)
}
}
// RegWebsocketConnectionOpenedEvent 注册连接打开事件
func (slf *websocketEvents) RegWebsocketConnectionOpenedEvent(handle WebsocketConnectionOpenedEventHandle) {
slf.websocketConnectionOpenedEventHandles = append(slf.websocketConnectionOpenedEventHandles, handle)
}
func (slf *websocketEvents) OnWebsocketConnectionOpenedEvent(conn *Websocket) {
for _, handle := range slf.websocketConnectionOpenedEventHandles {
handle(conn)
}
}
// RegWebsocketConnectionReceivePacketEvent 注册连接接收数据包事件
func (slf *websocketEvents) RegWebsocketConnectionReceivePacketEvent(handle WebsocketConnectionReceivePacketEventHandle) {
slf.websocketConnectionReceivePacketEventHandles = append(slf.websocketConnectionReceivePacketEventHandles, handle)
}
func (slf *websocketEvents) OnWebsocketConnectionReceivePacketEvent(conn *Websocket, packet server.Packet) {
for _, handle := range slf.websocketConnectionReceivePacketEventHandles {
handle(conn, packet)
}
}

141
server/conn.go
View File

@ -1,12 +1,14 @@
package server
import (
"context"
"github.com/gorilla/websocket"
"github.com/kercylan98/minotaur/utils/concurrent"
"github.com/kercylan98/minotaur/utils/super"
"github.com/kercylan98/minotaur/utils/log"
"github.com/panjf2000/gnet"
"github.com/xtaci/kcp-go/v5"
"net"
"runtime/debug"
"strings"
"sync"
"time"
@ -15,11 +17,14 @@ import (
// newKcpConn 创建一个处理KCP的连接
func newKcpConn(server *Server, session *kcp.UDPSession) *Conn {
c := &Conn{
server: server,
remoteAddr: session.RemoteAddr(),
ip: session.RemoteAddr().String(),
kcp: session,
data: map[any]any{},
ctx: server.ctx,
connection: &connection{
server: server,
remoteAddr: session.RemoteAddr(),
ip: session.RemoteAddr().String(),
kcp: session,
data: map[any]any{},
},
}
if index := strings.LastIndex(c.ip, ":"); index != -1 {
c.ip = c.ip[0:index]
@ -34,11 +39,14 @@ func newKcpConn(server *Server, session *kcp.UDPSession) *Conn {
// newKcpConn 创建一个处理GNet的连接
func newGNetConn(server *Server, conn gnet.Conn) *Conn {
c := &Conn{
server: server,
remoteAddr: conn.RemoteAddr(),
ip: conn.RemoteAddr().String(),
gn: conn,
data: map[any]any{},
ctx: server.ctx,
connection: &connection{
server: server,
remoteAddr: conn.RemoteAddr(),
ip: conn.RemoteAddr().String(),
gn: conn,
data: map[any]any{},
},
}
if index := strings.LastIndex(c.ip, ":"); index != -1 {
c.ip = c.ip[0:index]
@ -53,11 +61,14 @@ func newGNetConn(server *Server, conn gnet.Conn) *Conn {
// newKcpConn 创建一个处理WebSocket的连接
func newWebsocketConn(server *Server, ws *websocket.Conn, ip string) *Conn {
c := &Conn{
server: server,
remoteAddr: ws.RemoteAddr(),
ip: ip,
ws: ws,
data: map[any]any{},
ctx: server.ctx,
connection: &connection{
server: server,
remoteAddr: ws.RemoteAddr(),
ip: ip,
ws: ws,
data: map[any]any{},
},
}
var wait = new(sync.WaitGroup)
wait.Add(1)
@ -69,18 +80,13 @@ func newWebsocketConn(server *Server, ws *websocket.Conn, ip string) *Conn {
// newGatewayConn 创建一个处理网关消息的连接
func newGatewayConn(conn *Conn, connId string) *Conn {
c := &Conn{
server: conn.server,
data: map[any]any{},
//ctx: server.ctx,
connection: &connection{
server: conn.server,
data: map[any]any{},
},
}
c.gw = func(packet Packet) {
var gp = GP{
C: connId,
WT: packet.WebsocketType,
D: packet.Data,
T: time.Now().UnixNano(),
}
pd := super.MarshalJSON(&gp)
packet.Data = append(pd, 0xff)
c.gw = func(packet []byte) {
conn.Write(packet)
}
return c
@ -89,10 +95,13 @@ func newGatewayConn(conn *Conn, connId string) *Conn {
// NewEmptyConn 创建一个适用于测试的空连接
func NewEmptyConn(server *Server) *Conn {
c := &Conn{
server: server,
remoteAddr: &net.TCPAddr{},
ip: "0.0.0.0:0",
data: map[any]any{},
ctx: server.ctx,
connection: &connection{
server: server,
remoteAddr: &net.TCPAddr{},
ip: "0.0.0.0:0",
data: map[any]any{},
},
}
var wait = new(sync.WaitGroup)
wait.Add(1)
@ -101,15 +110,21 @@ func NewEmptyConn(server *Server) *Conn {
return c
}
// Conn 服务器连接
// Conn 服务器连接单次会话的包装
type Conn struct {
*connection
ctx context.Context
}
// connection 长久保持的连接
type connection struct {
server *Server
remoteAddr net.Addr
ip string
ws *websocket.Conn
gn gnet.Conn
kcp *kcp.UDPSession
gw func(packet Packet)
gw func(packet []byte)
data map[any]any
mutex sync.Mutex
packetPool *concurrent.Pool[*connPacket]
@ -174,7 +189,7 @@ func (slf *Conn) Close() {
slf.packets = nil
}
// SetData 设置连接数据
// SetData 设置连接数据，该数据将在连接关闭前始终存在
func (slf *Conn) SetData(key, value any) *Conn {
slf.data[key] = value
return slf
@ -185,6 +200,17 @@ func (slf *Conn) GetData(key any) any {
return slf.data[key]
}
// SetMessageData 设置消息数据，该数据将在消息处理完成后释放
func (slf *Conn) SetMessageData(key, value any) *Conn {
slf.ctx = context.WithValue(slf.ctx, key, value)
return slf
}
// GetMessageData 获取消息数据
func (slf *Conn) GetMessageData(key any) any {
return slf.ctx.Value(key)
}
// ReleaseData 释放数据
func (slf *Conn) ReleaseData() *Conn {
for k := range slf.data {
@ -198,28 +224,21 @@ func (slf *Conn) IsWebsocket() bool {
return slf.server.network == NetworkWebsocket
}
// Write 向连接中写入数据
// - messageType: websocket模式中指定消息类型
func (slf *Conn) Write(packet Packet) {
if slf.gw != nil {
slf.gw(packet)
return
}
packet = slf.server.OnConnectionWritePacketBeforeEvent(slf, packet)
if slf.packetPool == nil {
return
}
cp := slf.packetPool.Get()
cp.websocketMessageType = packet.WebsocketType
cp.packet = packet.Data
slf.mutex.Lock()
slf.packets = append(slf.packets, cp)
slf.mutex.Unlock()
// GetWST 获取websocket消息类型
func (slf *Conn) GetWST() int {
wst, _ := slf.ctx.Value(contextKeyWST).(int)
return wst
}
// WriteWithCallback 与 Write 相同，但是会在写入完成后调用 callback
// - 当 callback 为 nil 时，与 Write 相同
func (slf *Conn) WriteWithCallback(packet Packet, callback func(err error)) {
// SetWST 设置websocket消息类型
func (slf *Conn) SetWST(wst int) *Conn {
slf.ctx = context.WithValue(slf.ctx, contextKeyWST, wst)
return slf
}
// Write 向连接中写入数据
// - messageType: websocket模式中指定消息类型
func (slf *Conn) Write(packet []byte, callback ...func(err error)) {
if slf.gw != nil {
slf.gw(packet)
return
@ -229,9 +248,11 @@ func (slf *Conn) WriteWithCallback(packet Packet, callback func(err error)) {
return
}
cp := slf.packetPool.Get()
cp.websocketMessageType = packet.WebsocketType
cp.packet = packet.Data
cp.callback = callback
cp.wst = slf.GetWST()
cp.packet = packet
if len(callback) > 0 {
cp.callback = callback[0]
}
slf.mutex.Lock()
slf.packets = append(slf.packets, cp)
slf.mutex.Unlock()
@ -243,14 +264,16 @@ func (slf *Conn) writeLoop(wait *sync.WaitGroup) {
func() *connPacket {
return &connPacket{}
}, func(data *connPacket) {
data.wst = 0
data.packet = nil
data.websocketMessageType = 0
data.callback = nil
},
)
defer func() {
if err := recover(); err != nil {
slf.Close()
log.Error("WriteLoop", log.Any("Error", err))
debug.PrintStack()
}
}()
wait.Done()
@ -271,7 +294,7 @@ func (slf *Conn) writeLoop(wait *sync.WaitGroup) {
data := packets[i]
var err error
if slf.IsWebsocket() {
err = slf.ws.WriteMessage(data.websocketMessageType, data.packet)
err = slf.ws.WriteMessage(data.wst, data.packet)
} else {
if slf.gn != nil {
switch slf.server.network {

6
server/conn_packet.go
View File

@ -2,7 +2,7 @@ package server
// connPacket 连接包
type connPacket struct {
websocketMessageType int // websocket 消息类型
packet []byte // 数据包
callback func(err error) // 回调函数
wst int // websocket消息类型
packet []byte // 数据包
callback func(err error) // 回调函数
}

4
server/constants.go
View File

@ -26,3 +26,7 @@ const (
DefaultAsyncPoolSize = 256
DefaultWebsocketReadDeadline = 30 * time.Second
)
const (
contextKeyWST = "_wst" // WebSocket 消息类型
)

8
server/event.go
View File

@ -14,7 +14,7 @@ import (
type StartBeforeEventHandle func(srv *Server)
type StartFinishEventHandle func(srv *Server)
type StopEventHandle func(srv *Server)
type ConnectionReceivePacketEventHandle func(srv *Server, conn *Conn, packet Packet)
type ConnectionReceivePacketEventHandle func(srv *Server, conn *Conn, packet []byte)
type ConnectionOpenedEventHandle func(srv *Server, conn *Conn)
type ConnectionClosedEventHandle func(srv *Server, conn *Conn, err any)
type ReceiveCrossPacketEventHandle func(srv *Server, senderServerId int64, packet []byte)
@ -22,7 +22,7 @@ type MessageErrorEventHandle func(srv *Server, message *Message, err error)
type MessageLowExecEventHandle func(srv *Server, message *Message, cost time.Duration)
type ConsoleCommandEventHandle func(srv *Server)
type ConnectionOpenedAfterEventHandle func(srv *Server, conn *Conn)
type ConnectionWritePacketBeforeEventHandle func(srv *Server, conn *Conn, packet Packet) Packet
type ConnectionWritePacketBeforeEventHandle func(srv *Server, conn *Conn, packet []byte) []byte
type ShuntChannelCreatedEventHandle func(srv *Server, guid int64)
type ShuntChannelClosedEventHandle func(srv *Server, guid int64)
type ConnectionPacketPreprocessEventHandle func(srv *Server, conn *Conn, packet []byte, abort func(), usePacket func(newPacket []byte))
@ -201,7 +201,7 @@ func (slf *event) RegConnectionReceivePacketEvent(handle ConnectionReceivePacket
log.Info("Server", log.String("RegEvent", runtimes.CurrentRunningFuncName()), log.String("handle", reflect.TypeOf(handle).String()))
}
func (slf *event) OnConnectionReceivePacketEvent(conn *Conn, packet Packet) {
func (slf *event) OnConnectionReceivePacketEvent(conn *Conn, packet []byte) {
slf.connectionReceivePacketEventHandles.RangeValue(func(index int, value ConnectionReceivePacketEventHandle) bool {
value(slf.Server, conn, packet)
return true
@ -278,7 +278,7 @@ func (slf *event) RegConnectionWritePacketBeforeEvent(handle ConnectionWritePack
log.Info("Server", log.String("RegEvent", runtimes.CurrentRunningFuncName()), log.String("handle", reflect.TypeOf(handle).String()))
}
func (slf *event) OnConnectionWritePacketBeforeEvent(conn *Conn, packet Packet) (newPacket Packet) {
func (slf *event) OnConnectionWritePacketBeforeEvent(conn *Conn, packet []byte) (newPacket []byte) {
if slf.connectionWritePacketBeforeHandles.Len() == 0 {
return packet
}

68
server/gateway/endpoint.go
View File

@ -1,18 +1,20 @@
package gateway
import (
"github.com/alphadose/haxmap"
"github.com/kercylan98/minotaur/server"
"github.com/kercylan98/minotaur/server/client"
"github.com/kercylan98/minotaur/utils/super"
"time"
)
// NewEndpoint 创建网关端点
func NewEndpoint(name, address string, options ...EndpointOption) *Endpoint {
func NewEndpoint(gateway *Gateway, name string, client *client.Client, options ...EndpointOption) *Endpoint {
endpoint := &Endpoint{
client: client.NewWebsocket(address),
name: name,
address: address,
gateway: gateway,
client: client,
name: name,
address: client.GetServerAddr(),
connections: haxmap.New[string, *server.Conn](),
}
for _, option := range options {
option(endpoint)
@ -22,19 +24,37 @@ func NewEndpoint(name, address string, options ...EndpointOption) *Endpoint {
return 1 / (1 + 1.5*time.Duration(costUnixNano).Seconds())
}
}
endpoint.client.RegWebsocketConnectionClosedEvent(endpoint.onConnectionClosed)
endpoint.client.RegWebsocketConnectionReceivePacketEvent(endpoint.onConnectionReceivePacket)
endpoint.client.RegConnectionClosedEvent(endpoint.onConnectionClosed)
endpoint.client.RegConnectionReceivePacketEvent(endpoint.onConnectionReceivePacket)
return endpoint
}
// Endpoint 网关端点
type Endpoint struct {
client *client.Websocket // 端点客户端
name string // 端点名称
address string // 端点地址
state float64 // 端点健康值（0为不可用，越高越优）
offline bool // 离线
evaluator func(costUnixNano float64) float64 // 端点健康值评估函数
gateway *Gateway
client *client.Client // 端点客户端
name string // 端点名称
address string // 端点地址
state float64 // 端点健康值（0为不可用，越高越优）
offline bool // 离线
evaluator func(costUnixNano float64) float64 // 端点健康值评估函数
connections *haxmap.Map[string, *server.Conn] // 连接列表
}
// Link 连接端点
func (slf *Endpoint) Link(conn *server.Conn) {
slf.connections.Set(conn.GetID(), conn)
}
// Unlink 断开连接
func (slf *Endpoint) Unlink(conn *server.Conn) {
slf.connections.Del(conn.GetID())
}
// GetLink 获取连接
func (slf *Endpoint) GetLink(id string) *server.Conn {
conn, _ := slf.connections.Get(id)
return conn
}
// Offline 离线
@ -55,24 +75,28 @@ func (slf *Endpoint) Connect() {
}
// Write 写入数据
func (slf *Endpoint) Write(packet server.Packet) {
slf.client.Write(packet)
func (slf *Endpoint) Write(packet []byte, callback ...func(err error)) {
slf.client.Write(packet, callback...)
}
// WriteWS 写入 websocket 数据
func (slf *Endpoint) WriteWS(wst int, packet []byte, callback ...func(err error)) {
slf.client.WriteWS(wst, packet, callback...)
}
// onConnectionClosed 与端点连接断开事件
func (slf *Endpoint) onConnectionClosed(conn *client.Websocket, err any) {
func (slf *Endpoint) onConnectionClosed(conn *client.Client, err any) {
if !slf.offline {
go slf.Connect()
}
}
// onConnectionReceivePacket 接收到来自端点的数据包事件
func (slf *Endpoint) onConnectionReceivePacket(conn *client.Websocket, packet server.Packet) {
var gp server.GP
if err := super.UnmarshalJSON(packet.Data[:len(packet.Data)-1], &gp); err != nil {
func (slf *Endpoint) onConnectionReceivePacket(conn *client.Client, wst int, packet []byte) {
addr, sendTime, packet, err := UnmarshalGatewayInPacket(packet)
if err != nil {
panic(err)
}
cur := time.Now().UnixNano()
slf.state = slf.evaluator(float64(cur - gp.T))
conn.GetData(gp.C).(*server.Conn).Write(server.NewWSPacket(gp.WT, gp.D))
slf.state = slf.evaluator(float64(time.Now().UnixNano() - sendTime))
slf.GetLink(addr).SetWST(wst).Write(packet)
}

14
server/gateway/endpoint_manager.go
View File

@ -1,7 +1,7 @@
package gateway
import (
"github.com/kercylan98/minotaur/server"
"github.com/alphadose/haxmap"
"github.com/kercylan98/minotaur/utils/concurrent"
"github.com/kercylan98/minotaur/utils/random"
)
@ -10,7 +10,7 @@ import (
func NewEndpointManager() *EndpointManager {
em := &EndpointManager{
endpoints: concurrent.NewBalanceMap[string, []*Endpoint](),
memory: concurrent.NewBalanceMap[string, *Endpoint](),
memory: haxmap.New[string, *Endpoint](),
selector: func(endpoints []*Endpoint) *Endpoint {
return endpoints[random.Int(0, len(endpoints)-1)]
},
@ -21,13 +21,15 @@ func NewEndpointManager() *EndpointManager {
// EndpointManager 网关端点管理器
type EndpointManager struct {
endpoints *concurrent.BalanceMap[string, []*Endpoint]
memory *concurrent.BalanceMap[string, *Endpoint]
memory *haxmap.Map[string, *Endpoint]
selector func([]*Endpoint) *Endpoint
}
// GetEndpoint 获取端点
func (slf *EndpointManager) GetEndpoint(name string, conn *server.Conn) (*Endpoint, error) {
endpoint, exist := slf.memory.GetExist(conn.GetID())
// - name: 端点名称
// - id: 使用端点的连接标识
func (slf *EndpointManager) GetEndpoint(name, id string) (*Endpoint, error) {
endpoint, exist := slf.memory.Get(id)
if exist {
return endpoint, nil
}
@ -53,7 +55,7 @@ func (slf *EndpointManager) GetEndpoint(name string, conn *server.Conn) (*Endpoi
if endpoint == nil {
return nil, ErrEndpointNotExists
}
slf.memory.Set(conn.GetID(), endpoint)
slf.memory.Set(id, endpoint)
return endpoint, nil
}

29
server/gateway/gateway.go
View File

@ -2,7 +2,6 @@ package gateway
import (
"github.com/kercylan98/minotaur/server"
"github.com/kercylan98/minotaur/utils/super"
"math"
)
@ -36,28 +35,30 @@ func (slf *Gateway) Shutdown() {
slf.srv.Shutdown()
}
// onConnectionOpened 连接打开事件
func (slf *Gateway) onConnectionOpened(srv *server.Server, conn *server.Conn) {
endpoint, err := slf.GetEndpoint("test", conn)
endpoint, err := slf.GetEndpoint("test", conn.GetID())
if err != nil {
conn.Close()
return
}
endpoint.client.SetData(conn.GetID(), conn)
conn.SetData("endpoint", endpoint)
endpoint.Link(conn)
}
// onConnectionReceivePacket 连接接收数据包事件
func (slf *Gateway) onConnectionReceivePacket(srv *server.Server, conn *server.Conn, packet server.Packet) {
var gp = server.GP{
C: conn.GetID(),
WT: packet.WebsocketType,
D: packet.Data,
func (slf *Gateway) onConnectionReceivePacket(srv *server.Server, conn *server.Conn, packet []byte) {
endpoint, err := slf.GetEndpoint("test", conn.GetID())
if err != nil {
conn.Close()
return
}
pd := super.MarshalJSON(&gp)
packet.Data = append(pd, 0xff)
var endpoint, exist = conn.GetData("endpoint").(*Endpoint)
if exist {
packet, err = MarshalGatewayOutPacket(conn.GetID(), packet)
if err != nil {
conn.Close()
return
}
if conn.IsWebsocket() {
endpoint.WriteWS(conn.GetWST(), packet)
} else {
endpoint.Write(packet)
}
}

34
server/gateway/gateway_test.go
View File

@ -3,15 +3,39 @@ package gateway_test
import (
"fmt"
"github.com/kercylan98/minotaur/server"
gateway2 "github.com/kercylan98/minotaur/server/gateway"
"github.com/kercylan98/minotaur/server/client"
"github.com/kercylan98/minotaur/server/gateway"
"testing"
"time"
)
func TestGateway_RunEndpointServer(t *testing.T) {
srv := server.New(server.NetworkWebsocket, server.WithDeadlockDetect(time.Second*3))
srv.RegConnectionReceivePacketEvent(func(srv *server.Server, conn *server.Conn, packet server.Packet) {
fmt.Println("endpoint receive packet", string(packet.Data))
srv.RegConnectionClosedEvent(func(srv *server.Server, conn *server.Conn, err any) {
fmt.Println(err)
})
srv.RegConnectionPacketPreprocessEvent(func(srv *server.Server, conn *server.Conn, packet []byte, abort func(), usePacket func(newPacket []byte)) {
addr, packet, err := gateway.UnmarshalGatewayOutPacket(packet)
if err != nil {
// 非网关的普通数据包
return
}
usePacket(packet)
conn.SetMessageData("gw-addr", addr)
})
srv.RegConnectionWritePacketBeforeEvent(func(srv *server.Server, conn *server.Conn, packet []byte) []byte {
addr, ok := conn.GetMessageData("gw-addr").(string)
if !ok {
return packet
}
packet, err := gateway.MarshalGatewayInPacket(addr, time.Now().Unix(), packet)
if err != nil {
panic(err)
}
return packet
})
srv.RegConnectionReceivePacketEvent(func(srv *server.Server, conn *server.Conn, packet []byte) {
fmt.Println("endpoint receive packet", string(packet))
conn.Write(packet)
})
if err := srv.Run(":8889"); err != nil {
@ -21,9 +45,9 @@ func TestGateway_RunEndpointServer(t *testing.T) {
func TestGateway_Run(t *testing.T) {
srv := server.New(server.NetworkWebsocket, server.WithDeadlockDetect(time.Second*3))
gw := gateway2.NewGateway(srv)
gw := gateway.NewGateway(srv)
srv.RegStartFinishEvent(func(srv *server.Server) {
if err := gw.AddEndpoint(gateway2.NewEndpoint("test", "ws://127.0.0.1:8889")); err != nil {
if err := gw.AddEndpoint(gateway.NewEndpoint(gw, "test", client.NewWebsocket("ws://127.0.0.1:8889"))); err != nil {
panic(err)
}
})

109
server/gateway/packet.go
View File

@ -1,7 +1,108 @@
package gateway
type Packet struct {
ConnID string
WebsocketType int
Data []byte
import (
"encoding/binary"
"errors"
"fmt"
"net"
"strconv"
)
var packetIdentifier = []byte{0xDE, 0xAD, 0xBE, 0xEF}
// MarshalGatewayOutPacket 将数据包转换为网关出网数据包
// - | identifier(4) | ipv4(4) | port(2) | packet |
func MarshalGatewayOutPacket(addr string, packet []byte) ([]byte, error) {
host, portStr, err := net.SplitHostPort(addr)
if err != nil {
return nil, err
}
ipBytes := net.ParseIP(host).To4()
if ipBytes == nil {
return nil, errors.New("invalid IPv4 address")
}
port, err := strconv.Atoi(portStr)
if err != nil || port < 0 || port > 65535 {
return nil, errors.New("invalid port number")
}
portBytes := []byte{byte(port >> 8), byte(port & 0xFF)}
result := append(packetIdentifier, ipBytes...)
result = append(result, portBytes...)
result = append(result, packet...)
return result, nil
}
// UnmarshalGatewayOutPacket 将网关出网数据包转换为数据包
// - | identifier(4) | ipv4(4) | port(2) | packet |
func UnmarshalGatewayOutPacket(data []byte) (addr string, packet []byte, err error) {
if len(data) < 10 {
err = errors.New("data is too short to contain an IPv4 address and a port")
return
}
if !compareBytes(data[:4], packetIdentifier) {
err = errors.New("invalid identifier")
return
}
ipAddr := net.IP(data[4:8]).String()
port := uint16(data[8])<<8 | uint16(data[9])
addr = fmt.Sprintf("%s:%d", ipAddr, port)
packet = data[10:]
return addr, packet, nil
}
// MarshalGatewayInPacket 将数据包转换为网关入网数据包
// - | ipv4(4) | port(2) | cost(4) | packet |
func MarshalGatewayInPacket(addr string, currentTime int64, packet []byte) ([]byte, error) {
host, portStr, err := net.SplitHostPort(addr)
if err != nil {
return nil, err
}
ipBytes := net.ParseIP(host).To4()
if ipBytes == nil {
return nil, errors.New("invalid IPv4 address")
}
port, err := strconv.Atoi(portStr)
if err != nil || port < 0 || port > 65535 {
return nil, errors.New("invalid port number")
}
portBytes := []byte{byte(port >> 8), byte(port & 0xFF)}
costBytes := make([]byte, 4)
binary.BigEndian.PutUint32(costBytes, uint32(currentTime))
result := append(ipBytes, portBytes...)
result = append(result, costBytes...)
result = append(result, packet...)
return result, nil
}
// UnmarshalGatewayInPacket 将网关入网数据包转换为数据包
// - | ipv4(4) | port(2) | cost(4) | packet |
func UnmarshalGatewayInPacket(data []byte) (addr string, sendTime int64, packet []byte, err error) {
if len(data) < 10 {
err = errors.New("data is too short")
return
}
ipAddr := net.IP(data[:4]).String()
port := uint16(data[4])<<8 | uint16(data[5])
addr = fmt.Sprintf("%s:%d", ipAddr, port)
sendTime = int64(binary.BigEndian.Uint32(data[6:10]))
packet = data[10:]
return addr, sendTime, packet, nil
}
func compareBytes(a, b []byte) bool {
if len(a) != len(b) {
return false
}
for i := range a {
if a[i] != b[i] {
return false
}
}
return true
}

2
server/gnet.go
View File

@ -39,7 +39,7 @@ func (slf *gNet) AfterWrite(c gnet.Conn, b []byte) {
}
func (slf *gNet) React(packet []byte, c gnet.Conn) (out []byte, action gnet.Action) {
PushPacketMessage(slf.Server, c.Context().(*Conn), append(bytes.Clone(packet), 0))
PushPacketMessage(slf.Server, c.Context().(*Conn), 0, append(bytes.Clone(packet), 0))
return nil, gnet.None
}

8
server/gp.go
View File

@ -1,8 +0,0 @@
package server
type GP struct {
C string // 连接 ID
WT int // WebSocket 类型
D []byte // 数据
T int64 // 时间戳
}

4
server/lockstep/client.go
View File

@ -1,12 +1,10 @@
package lockstep
import "github.com/kercylan98/minotaur/server"
// Client 帧同步客户端接口定义
// - 客户端应该具备ID及写入数据包的实现
type Client[ID comparable] interface {
// GetID 用户玩家ID
GetID() ID
// Write 写入数据包
Write(packet server.Packet)
Write(packet []byte, callback ...func(err error))
}

3
server/lockstep/evnets.go Normal file
View File

@ -0,0 +1,3 @@
package lockstep
type StoppedEventHandle[ClientID comparable, Command any] func(lockstep *Lockstep[ClientID, Command])

16
server/lockstep/lockstep.go
View File

@ -2,8 +2,6 @@ package lockstep
import (
"encoding/json"
"github.com/kercylan98/minotaur/component"
"github.com/kercylan98/minotaur/server"
"github.com/kercylan98/minotaur/utils/concurrent"
"github.com/kercylan98/minotaur/utils/timer"
"sync"
@ -18,13 +16,13 @@ func NewLockstep[ClientID comparable, Command any](options ...Option[ClientID, C
frames: concurrent.NewBalanceMap[int, []Command](),
ticker: timer.GetTicker(10),
frameRate: 15,
serialization: func(frame int, commands []Command) server.Packet {
serialization: func(frame int, commands []Command) []byte {
frameStruct := struct {
Frame int `json:"frame"`
Commands []Command `json:"commands"`
}{frame, commands}
data, _ := json.Marshal(frameStruct)
return server.NewPacket(data)
return data
},
clientCurrentFrame: concurrent.NewBalanceMap[ClientID, int](),
}
@ -49,11 +47,11 @@ type Lockstep[ClientID comparable, Command any] struct {
clientCurrentFrame *concurrent.BalanceMap[ClientID, int] // 客户端当前帧数
running atomic.Bool
frameRate int // 帧率（每秒N帧）
frameLimit int // 帧上限
serialization func(frame int, commands []Command) server.Packet // 序列化函数
frameRate int // 帧率（每秒N帧）
frameLimit int // 帧上限
serialization func(frame int, commands []Command) []byte // 序列化函数
lockstepStoppedEventHandles []component.LockstepStoppedEventHandle[ClientID, Command]
lockstepStoppedEventHandles []StoppedEventHandle[ClientID, Command]
}
// JoinClient 加入客户端到广播队列中
@ -156,7 +154,7 @@ func (slf *Lockstep[ClientID, Command]) GetFrames() [][]Command {
}
// RegLockstepStoppedEvent 当广播停止时将触发被注册的事件处理函数
func (slf *Lockstep[ClientID, Command]) RegLockstepStoppedEvent(handle component.LockstepStoppedEventHandle[ClientID, Command]) {
func (slf *Lockstep[ClientID, Command]) RegLockstepStoppedEvent(handle StoppedEventHandle[ClientID, Command]) {
slf.lockstepStoppedEventHandles = append(slf.lockstepStoppedEventHandles, handle)
}

4
server/lockstep/lockstep_options.go
View File

@ -1,7 +1,5 @@
package lockstep
import "github.com/kercylan98/minotaur/server"
type Option[ClientID comparable, Command any] func(lockstep *Lockstep[ClientID, Command])
// WithFrameLimit 通过特定逻辑帧上限创建锁步（帧）同步组件
@ -31,7 +29,7 @@ func WithFrameRate[ClientID comparable, Command any](frameRate int) Option[Clien
// Frame int `json:"frame"`
// Commands []Command `json:"commands"`
// }
func WithSerialization[ClientID comparable, Command any](handle func(frame int, commands []Command) server.Packet) Option[ClientID, Command] {
func WithSerialization[ClientID comparable, Command any](handle func(frame int, commands []Command) []byte) Option[ClientID, Command] {
return func(lockstep *Lockstep[ClientID, Command]) {
lockstep.serialization = handle
}

5
server/message.go
View File

@ -1,6 +1,7 @@
package server
import (
"context"
"encoding/json"
"fmt"
"reflect"
@ -104,10 +105,10 @@ func (slf MessageType) String() string {
}
// PushPacketMessage 向特定服务器中推送 MessageTypePacket 消息
func PushPacketMessage(srv *Server, conn *Conn, packet []byte, mark ...any) {
func PushPacketMessage(srv *Server, conn *Conn, wst int, packet []byte, mark ...any) {
msg := srv.messagePool.Get()
msg.t = MessageTypePacket
msg.attrs = append([]any{conn, packet}, mark...)
msg.attrs = append([]any{&Conn{ctx: context.WithValue(conn.ctx, contextKeyWST, wst), connection: conn.connection}, packet}, mark...)
srv.pushMessage(msg)
}

42
server/packet.go
View File

@ -1,42 +0,0 @@
package server
// NewPacket 创建一个数据包
func NewPacket(data []byte) Packet {
return Packet{
Data: data,
}
}
// NewWSPacket 创建一个 websocket 数据包
func NewWSPacket(websocketType int, data []byte) Packet {
return Packet{
WebsocketType: websocketType,
Data: data,
}
}
// NewPacketString 创建一个字符串数据包
func NewPacketString(data string) Packet {
return Packet{
Data: []byte(data),
}
}
// NewWSPacketString 创建一个 websocket 字符串数据包
func NewWSPacketString(websocketType int, data string) Packet {
return Packet{
WebsocketType: websocketType,
Data: []byte(data),
}
}
// Packet 数据包
type Packet struct {
WebsocketType int // websocket 消息类型
Data []byte // 数据
}
// String 转换为字符串
func (slf Packet) String() string {
return string(slf.Data)
}

25
server/server.go
View File

@ -36,6 +36,7 @@ func New(network Network, options ...Option) *Server {
online: concurrent.NewBalanceMap[string, *Conn](),
closeChannel: make(chan struct{}, 1),
systemSignal: make(chan os.Signal, 1),
ctx: context.Background(),
}
server.event = newEvent(server)
@ -96,6 +97,7 @@ type Server struct {
channelGenerator func(guid int64) chan *Message // 消息管道生成器
shuntMatcher func(conn *Conn) (guid int64, allowToCreate bool) // 分流管道匹配器
messageCounter atomic.Int64 // 消息计数器
ctx context.Context // 上下文
}
// Run 使用特定地址运行服务器
@ -212,7 +214,7 @@ func (slf *Server) Run(addr string) error {
if err != nil {
panic(err)
}
PushPacketMessage(slf, conn, buf[:n])
PushPacketMessage(slf, conn, 0, buf[:n])
}
}(conn)
}
@ -303,7 +305,7 @@ func (slf *Server) Run(addr string) error {
if len(slf.supportMessageTypes) > 0 && !slf.supportMessageTypes[messageType] {
panic(ErrWebsocketIllegalMessageType)
}
PushPacketMessage(slf, conn, append(packet, byte(messageType)))
PushPacketMessage(slf, conn, messageType, packet)
}
})
go func() {
@ -629,25 +631,8 @@ func (slf *Server) dispatchMessage(msg *Message) {
case MessageTypePacket:
var conn = attrs[0].(*Conn)
var packet = attrs[1].([]byte)
var wst = int(packet[len(packet)-1])
if len(packet) >= 2 {
var ct = packet[len(packet)-2]
if ct == 0xff {
var gp GP
if err := super.UnmarshalJSON(packet[:len(packet)-2], &gp); err != nil {
panic(err)
}
packet = gp.D
conn = newGatewayConn(conn, gp.C)
} else {
packet = packet[:len(packet)-1]
}
} else {
packet = packet[:len(packet)-1]
}
if !slf.OnConnectionPacketPreprocessEvent(conn, packet, func(newPacket []byte) { packet = newPacket }) {
slf.OnConnectionReceivePacketEvent(conn, Packet{Data: packet, WebsocketType: wst})
slf.OnConnectionReceivePacketEvent(conn, packet)
}
case MessageTypeError:
err, action := attrs[0].(error), attrs[1].(MessageErrorAction)

2
server/server_example_test.go
View File

@ -11,7 +11,7 @@ func ExampleNew() {
server.WithPProf("/debug/pprof"),
)
srv.RegConnectionReceivePacketEvent(func(srv *server.Server, conn *server.Conn, packet server.Packet) {
srv.RegConnectionReceivePacketEvent(func(srv *server.Server, conn *server.Conn, packet []byte) {
conn.Write(packet)
})

4
utils/stream/slice.go
View File

@ -51,7 +51,7 @@ func (slf Slice[V]) RandomKeep(n int) Slice[V] {
if n >= length {
return slf
}
var hit = make([]int, length, length)
var hit = make([]int, length)
for i := 0; i < n; i++ {
hit[i] = 1
}
@ -71,7 +71,7 @@ func (slf Slice[V]) RandomDelete(n int) Slice[V] {
if n >= length {
return slf[:0]
}
var hit = make([]int, length, length)
var hit = make([]int, length)
for i := 0; i < n; i++ {
hit[i] = 1
}

2
utils/super/stack.go
View File

@ -20,7 +20,7 @@ type StackGo struct {
// Wait 等待收集消息堆栈
// - 在调用 Wait 函数后，当前协程将会被挂起，直到调用 Stack 或 GiveUp 函数
func (slf *StackGo) Wait() {
slf.stack = make(chan *struct{}, 0)
slf.stack = make(chan *struct{})
if s := <-slf.stack; s != nil {
slf.collect <- debug.Stack()
}