other: v2 server 增加 kcp 网络支持

internal/server/client/client_test.go

@@ -1,34 +1,10 @@
 package client_test
 
 import (
-	"github.com/kercylan98/minotaur/server"
-	"github.com/kercylan98/minotaur/server/client"
-	"sync"
+	"github.com/kercylan98/minotaur/internal/server/client"
 	"testing"
 )
 
 func TestClient_WriteWS(t *testing.T) {
-	var wait sync.WaitGroup
-	wait.Add(1)
-	srv := server.New(server.NetworkWebsocket)
-	srv.RegConnectionReceivePacketEvent(func(srv *server.Server, conn *server.Conn, packet []byte) {
-		srv.Shutdown()
-	})
-	srv.RegStopEvent(func(srv *server.Server) {
-		wait.Done()
-	})
-	srv.RegMessageReadyEvent(func(srv *server.Server) {
-		cli := client.NewWebsocket("ws://127.0.0.1:9999")
-		cli.RegConnectionOpenedEvent(func(conn *client.Client) {
-			conn.WriteWS(2, []byte("Hello"))
-		})
-		if err := cli.Run(); err != nil {
-			panic(err)
-		}
-	})
-	if err := srv.Run(":9999"); err != nil {
-		panic(err)
-	}
-
-	wait.Wait()
+	client.NewWebsocket("ws://127.0.0.1:9999")
 }

server/conn.go

@@ -97,6 +97,9 @@ type Conn interface {
 
 	// Close 关闭连接
 	Close()
+
+	// IsClosed 是否已关闭
+	IsClosed() bool
 }
 
 func newConn(srv *server, c net.Conn, connWriter ConnWriter) *conn {
@@ -289,3 +292,7 @@ func (c *conn) Close() {
 	// 清理资源
 	_ = c.conn.Close()
 }
+
+func (c *conn) IsClosed() bool {
+	return atomic.LoadInt32(&c.state) == ConnStatusClosed
+}

server/network/kcp.go

@@ -0,0 +1,103 @@
+package network
+
+import (
+	"context"
+	"github.com/kercylan98/minotaur/server"
+	"github.com/xtaci/kcp-go/v5"
+	"runtime"
+	"sync"
+	"sync/atomic"
+)
+
+var kcpInitOnce sync.Once
+
+func init() {
+	kcp.SystemTimedSched.Close() // 默认禁用 KCP 系统定时器
+	kcp.SystemTimedSched = nil
+}
+
+func newKcpCore(addr string) server.Network {
+	kcpInitOnce.Do(func() {
+		if kcp.SystemTimedSched == nil {
+			kcp.SystemTimedSched = kcp.NewTimedSched(runtime.NumCPU())
+		}
+	})
+	return &kcpCore{
+		addr: addr,
+	}
+}
+
+type kcpCore struct {
+	ctx        context.Context
+	controller server.Controller
+	addr       string
+	closed     atomic.Bool
+}
+
+func (k *kcpCore) OnSetup(ctx context.Context, controller server.Controller) error {
+	k.ctx = ctx
+	k.controller = controller
+	return nil
+}
+
+func (k *kcpCore) OnRun() error {
+	lis, err := kcp.ListenWithOptions(k.addr, nil, 0, 0)
+	if err != nil {
+		return err
+	}
+	defer func(lis *kcp.Listener) {
+		_ = lis.Close()
+	}(lis)
+	for !k.closed.Load() {
+		var conn *kcp.UDPSession
+		var srvConn server.Conn
+		if conn, err = lis.AcceptKCP(); err != nil {
+			continue
+		}
+
+		// 注册连接
+		k.controller.RegisterConnection(conn,
+			func(packet server.Packet) (err error) {
+				if _, err = conn.Write(packet.GetBytes()); err != nil {
+					k.controller.OnConnectionAsyncWriteError(srvConn, packet, err)
+				}
+				return
+			}, func(conn server.Conn) {
+				srvConn = conn
+			},
+		)
+
+		// 处理连接数据
+		go func(ctx context.Context, conn *kcp.UDPSession, srvConn server.Conn) {
+			var buf = make([]byte, 1024)
+			var n int
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				default:
+					if n, err = conn.Read(buf); err != nil {
+						srvConn.Close()
+						return
+					}
+					k.controller.ReactPacket(conn, server.NewPacket(buf[:n]))
+				}
+			}
+		}(k.ctx, conn, srvConn)
+	}
+
+	return nil
+}
+
+func (k *kcpCore) OnShutdown() error {
+	k.closed.Store(true)
+	return nil
+}
+
+func (k *kcpCore) Schema() string {
+	return "kcp(udp)"
+}
+
+func (k *kcpCore) Address() string {
+	return k.addr
+}

server/network/networks.go

@@ -72,3 +72,9 @@ func Udp6(addr string) server.Network {
 func Unix(addr string) server.Network {
 	return newGNetCore(new(gNetGenericHandler), schemaUnix, addr)
 }
+
+// Kcp 创建一个 KCP 网络
+//   - addr 期望为类似于 127.0.0.1:1234 或 :1234 的地址
+func Kcp(addr string) server.Network {
+	return newKcpCore(addr)
+}