feat(net): 协议编解码（帧/varint/消息）与服务器网络层（握手/心跳/半包处理，含测试）
This commit is contained in:
NianGao Dev
225
internal/netserver/netserver.go
Normal file
225
internal/netserver/netserver.go
Normal file
@@ -0,0 +1,225 @@
|
||||
// Package netserver 服务器网络层：TCP 监听、连接会话、握手与心跳（多人同步.md、网络协议.md §4–§7）。
|
||||
//
|
||||
// 线程模型（架构.md §4）：每连接 1 读 + 1 写 goroutine；业务回调在主逻辑线程执行（通过
|
||||
// 事件 channel 投递），tick 内不做 IO。
|
||||
package netserver
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"mc/internal/logx"
|
||||
"mc/internal/netproto"
|
||||
)
|
||||
|
||||
// 连接超时与心跳参数（网络协议.md §7）。
|
||||
const (
|
||||
readTimeout = 30 * time.Second // 无响应判定断线
|
||||
heartbeatIn = 5 * time.Second // 心跳间隔
|
||||
maxFrameSize = netproto.MaxFrameSize
|
||||
)
|
||||
|
||||
// Handler 服务器业务回调（由主逻辑线程实现）。
|
||||
type Handler interface {
|
||||
// OnConnect 连接建立（握手前）。
|
||||
OnConnect(c *Conn)
|
||||
// OnHandshake 收到握手：返回登录结果（0 成功，其他为拒绝原因码）。
|
||||
OnHandshake(c *Conn, hs netproto.Handshake) uint8
|
||||
// OnMove 玩家移动（20Hz）。
|
||||
OnMove(c *Conn, m netproto.PlayerMove)
|
||||
// OnDisconnect 连接断开。
|
||||
OnDisconnect(c *Conn)
|
||||
}
|
||||
|
||||
// Server TCP 服务器。
|
||||
type Server struct {
|
||||
addr string
|
||||
log *logx.Logger
|
||||
handler Handler
|
||||
|
||||
ln net.Listener
|
||||
mu sync.Mutex
|
||||
conns map[uint64]*Conn
|
||||
|
||||
nextID atomic.Uint64
|
||||
}
|
||||
|
||||
// New 创建服务器（未监听）。
|
||||
func New(addr string, log *logx.Logger, h Handler) *Server {
|
||||
return &Server{addr: addr, log: log, handler: h, conns: make(map[uint64]*Conn)}
|
||||
}
|
||||
|
||||
// Listen 绑定端口并启动接受循环（ctx 取消即关闭）。
|
||||
func (s *Server) Listen(ctx context.Context) error {
|
||||
ln, err := net.Listen("tcp", s.addr)
|
||||
if err != nil {
|
||||
return fmt.Errorf("netserver.Listen %s: %w", s.addr, err)
|
||||
}
|
||||
s.ln = ln
|
||||
s.log.Infof("网络监听于 %s", s.addr)
|
||||
go s.acceptLoop(ctx)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Addr 返回监听地址。
|
||||
func (s *Server) Addr() string { return s.ln.Addr().String() }
|
||||
|
||||
// acceptLoop 接受连接循环。
|
||||
func (s *Server) acceptLoop(ctx context.Context) {
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
s.Close()
|
||||
}()
|
||||
for {
|
||||
nc, err := s.ln.Accept()
|
||||
if err != nil {
|
||||
return // 关闭监听即退出
|
||||
}
|
||||
c := newConn(s.nextID.Add(1), nc, s)
|
||||
s.mu.Lock()
|
||||
s.conns[c.id] = c
|
||||
s.mu.Unlock()
|
||||
go c.readLoop()
|
||||
go c.writeLoop()
|
||||
}
|
||||
}
|
||||
|
||||
// Close 关闭监听与全部连接（优雅退出）。
|
||||
func (s *Server) Close() {
|
||||
if s.ln != nil {
|
||||
_ = s.ln.Close()
|
||||
}
|
||||
s.mu.Lock()
|
||||
conns := make([]*Conn, 0, len(s.conns))
|
||||
for _, c := range s.conns {
|
||||
conns = append(conns, c)
|
||||
}
|
||||
s.mu.Unlock()
|
||||
for _, c := range conns {
|
||||
c.close()
|
||||
}
|
||||
}
|
||||
|
||||
// Conn 一个客户端连接。
|
||||
type Conn struct {
|
||||
id uint64
|
||||
conn net.Conn
|
||||
srv *Server
|
||||
send chan []byte
|
||||
closeOnce sync.Once
|
||||
}
|
||||
|
||||
// newConn 创建连接。
|
||||
func newConn(id uint64, nc net.Conn, s *Server) *Conn {
|
||||
return &Conn{id: id, conn: nc, srv: s, send: make(chan []byte, 64)}
|
||||
}
|
||||
|
||||
// ID 连接 ID（服务器分配，多人同步.md §3）。
|
||||
func (c *Conn) ID() uint64 { return c.id }
|
||||
|
||||
// readLoop 读循环：解帧并分发（网络协议.md §1 帧格式）。
|
||||
// 半包（ErrIncomplete）时保留缓冲区等待更多数据；协议错误直接断开。
|
||||
func (c *Conn) readLoop() {
|
||||
defer c.srv.handler.OnDisconnect(c)
|
||||
buf := make([]byte, 0, 64*1024)
|
||||
tmp := make([]byte, 64*1024)
|
||||
for {
|
||||
_ = c.conn.SetReadDeadline(time.Now().Add(readTimeout))
|
||||
n, err := c.conn.Read(tmp)
|
||||
if err != nil {
|
||||
return // 超时/断开：结束连接
|
||||
}
|
||||
buf = append(buf, tmp[:n]...)
|
||||
c.srv.log.Debugf("连接 %d 读到 %d 字节（缓冲 %d）", c.id, n, len(buf))
|
||||
if len(buf) > 2*maxFrameSize {
|
||||
c.srv.log.Warnf("连接 %d 缓冲区超限，断开", c.id)
|
||||
c.close()
|
||||
return
|
||||
}
|
||||
for len(buf) >= 4 {
|
||||
f, rest, err := netproto.DecodeFrame(buf)
|
||||
if err != nil {
|
||||
if errors.Is(err, netproto.ErrIncomplete) {
|
||||
break // 半包：保留 buf，等待更多数据
|
||||
}
|
||||
c.srv.log.Warnf("连接 %d 帧解析失败: %v，断开", c.id, err)
|
||||
c.close()
|
||||
return
|
||||
}
|
||||
buf = rest
|
||||
c.handle(f)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// handle 消息分发（读 goroutine → 业务回调，业务侧自行保证线程安全）。
|
||||
func (c *Conn) handle(f netproto.Frame) {
|
||||
switch f.MsgID {
|
||||
case netproto.MsgHandshake:
|
||||
hs, err := netproto.DecodeHandshake(f.Payload)
|
||||
if err != nil {
|
||||
c.close()
|
||||
return
|
||||
}
|
||||
code := c.srv.handler.OnHandshake(c, hs)
|
||||
c.Send(netproto.Frame{MsgID: netproto.MsgLoginResponse, Payload: []byte{code}})
|
||||
if code != 0 {
|
||||
c.close()
|
||||
}
|
||||
case netproto.MsgPlayerMove:
|
||||
m, err := netproto.DecodePlayerMove(f.Payload)
|
||||
if err == nil {
|
||||
c.srv.handler.OnMove(c, m)
|
||||
}
|
||||
case netproto.MsgHeartbeat:
|
||||
c.Send(netproto.Frame{MsgID: netproto.MsgHeartbeat, Payload: f.Payload}) // 回显心跳
|
||||
case netproto.MsgDisconnect:
|
||||
c.close()
|
||||
}
|
||||
}
|
||||
|
||||
// writeLoop 写循环：channel 驱动，写超时断开（网络协议.md §7）。
|
||||
func (c *Conn) writeLoop() {
|
||||
for data := range c.send {
|
||||
_ = c.conn.SetWriteDeadline(time.Now().Add(readTimeout))
|
||||
if _, err := c.conn.Write(data); err != nil {
|
||||
c.close()
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Send 发送一帧（非阻塞入队；队满丢弃并告警，防慢连接阻塞服务器）。
|
||||
func (c *Conn) Send(f netproto.Frame) {
|
||||
select {
|
||||
case c.send <- netproto.EncodeFrame(f):
|
||||
default:
|
||||
c.srv.log.Warnf("连接 %d 发送队列已满，丢弃帧 0x%02X", c.id, f.MsgID)
|
||||
}
|
||||
}
|
||||
|
||||
// close 幂等关闭。
|
||||
func (c *Conn) close() {
|
||||
c.closeOnce.Do(func() {
|
||||
_ = c.conn.Close()
|
||||
close(c.send)
|
||||
c.srv.mu.Lock()
|
||||
delete(c.srv.conns, c.id)
|
||||
c.srv.mu.Unlock()
|
||||
})
|
||||
}
|
||||
|
||||
// Count 当前连接数（测试/统计）。
|
||||
func (s *Server) Count() int {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return len(s.conns)
|
||||
}
|
||||
|
||||
var _ = io.EOF // 保持 io 引用（后续扩展流式读取）
|
||||
126
internal/netserver/netserver_test.go
Normal file
126
internal/netserver/netserver_test.go
Normal file
@@ -0,0 +1,126 @@
|
||||
package netserver
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"mc/internal/logx"
|
||||
"mc/internal/netproto"
|
||||
)
|
||||
|
||||
// testHandler 记录回调的测试处理器。
|
||||
type testHandler struct {
|
||||
logins chan netproto.Handshake
|
||||
moves chan netproto.PlayerMove
|
||||
connects chan uint64
|
||||
}
|
||||
|
||||
func newTestHandler() *testHandler {
|
||||
return &testHandler{
|
||||
logins: make(chan netproto.Handshake, 8),
|
||||
moves: make(chan netproto.PlayerMove, 8),
|
||||
connects: make(chan uint64, 8),
|
||||
}
|
||||
}
|
||||
|
||||
func (h *testHandler) OnConnect(c *Conn) { h.connects <- c.ID() }
|
||||
func (h *testHandler) OnHandshake(c *Conn, hs netproto.Handshake) uint8 {
|
||||
h.logins <- hs
|
||||
return 0
|
||||
}
|
||||
func (h *testHandler) OnMove(c *Conn, m netproto.PlayerMove) { h.moves <- m }
|
||||
func (h *testHandler) OnDisconnect(c *Conn) {}
|
||||
|
||||
// startServer 启动测试服务器。
|
||||
func startServer(t *testing.T) (*Server, *testHandler, context.CancelFunc) {
|
||||
t.Helper()
|
||||
log, err := logx.New("", logx.LevelDebug) // 仅 stderr（避免临时文件句柄）
|
||||
if err != nil {
|
||||
t.Fatalf("创建日志失败: %v", err)
|
||||
}
|
||||
h := newTestHandler()
|
||||
s := New("127.0.0.1:0", log, h)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
if err := s.Listen(ctx); err != nil {
|
||||
t.Fatalf("监听失败: %v", err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
cancel()
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
})
|
||||
return s, h, cancel
|
||||
}
|
||||
|
||||
// dial 建立客户端连接。
|
||||
func dial(t *testing.T, addr string) net.Conn {
|
||||
t.Helper()
|
||||
c, err := net.DialTimeout("tcp", addr, 3*time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("拨号失败: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = c.Close() })
|
||||
return c
|
||||
}
|
||||
|
||||
// TestHandshakeFlow 握手 → 登录响应 → 心跳回显全链路（网络协议.md §4）。
|
||||
func TestHandshakeFlow(t *testing.T) {
|
||||
s, h, _ := startServer(t)
|
||||
nc := dial(t, s.Addr())
|
||||
|
||||
// 发送握手
|
||||
hs := netproto.Handshake{ProtocolVersion: 1, ClientID: "测试客户端"}
|
||||
_, err := nc.Write(netproto.EncodeFrame(netproto.Frame{MsgID: netproto.MsgHandshake, Payload: hs.Encode()}))
|
||||
if err != nil {
|
||||
t.Fatalf("发送握手失败: %v", err)
|
||||
}
|
||||
select {
|
||||
case got := <-h.logins:
|
||||
if got != hs {
|
||||
t.Fatalf("握手内容不一致: %+v", got)
|
||||
}
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("未收到握手回调")
|
||||
}
|
||||
|
||||
// 读登录响应帧
|
||||
_ = nc.SetReadDeadline(time.Now().Add(3 * time.Second))
|
||||
buf := make([]byte, 256)
|
||||
n, err := nc.Read(buf)
|
||||
if err != nil {
|
||||
t.Fatalf("读登录响应失败: %v", err)
|
||||
}
|
||||
f, _, err := netproto.DecodeFrame(buf[:n])
|
||||
if err != nil || f.MsgID != netproto.MsgLoginResponse {
|
||||
t.Fatalf("登录响应异常: %+v err=%v", f, err)
|
||||
}
|
||||
if len(f.Payload) != 1 || f.Payload[0] != 0 {
|
||||
t.Fatalf("登录结果应为 0（成功），实际 %v", f.Payload)
|
||||
}
|
||||
|
||||
// 心跳回显
|
||||
_, _ = nc.Write(netproto.EncodeFrame(netproto.Frame{MsgID: netproto.MsgHeartbeat, Payload: []byte{7}}))
|
||||
n, _ = nc.Read(buf)
|
||||
f, _, _ = netproto.DecodeFrame(buf[:n])
|
||||
if f.MsgID != netproto.MsgHeartbeat || len(f.Payload) != 1 || f.Payload[0] != 7 {
|
||||
t.Fatalf("心跳回显异常: %+v", f)
|
||||
}
|
||||
}
|
||||
|
||||
// TestMoveBroadcast 移动消息分发。
|
||||
func TestMoveBroadcast(t *testing.T) {
|
||||
s, h, _ := startServer(t)
|
||||
nc := dial(t, s.Addr())
|
||||
m := netproto.PlayerMove{X: 1.5, Y: 64, Z: -2, Yaw: 0.5, Pitch: 0, OnGround: true}
|
||||
_, _ = nc.Write(netproto.EncodeFrame(netproto.Frame{MsgID: netproto.MsgPlayerMove, Payload: m.Encode()}))
|
||||
select {
|
||||
case got := <-h.moves:
|
||||
if got != m {
|
||||
t.Fatalf("移动数据不一致: %+v", got)
|
||||
}
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("未收到移动回调")
|
||||
}
|
||||
_ = s.Count() // 连接计数可用
|
||||
}
|
||||
Reference in New Issue
Block a user