// Package netclient 客户端网络层:连接服务器、登录、接收区块/方块变更、发送移动(多人同步.md §2)。 // // 线程模型(架构.md §4):1 读 + 1 写 goroutine;业务回调经 Handler 在主逻辑线程消费。 package netclient import ( "fmt" "net" "sync" "sync/atomic" "time" "mc/internal/netproto" ) // 读写超时(网络协议.md §7)。 const ioTimeout = 30 * time.Second // Handler 客户端业务回调。 type Handler interface { // OnLogin 登录结果(0 成功)。 OnLogin(code uint8) // OnChunk 收到区块数据(save.EncodeChunk 格式,调用方 DecodeChunk)。 OnChunk(data []byte) // OnBlockChange 方块变更。 OnBlockChange(m netproto.BlockChange) // OnEntityMove 其他实体移动(多人同步.md §3)。 OnEntityMove(m netproto.EntityMove) } // Client TCP 客户端。 type Client struct { conn net.Conn send chan []byte done chan struct{} closed atomic.Bool closeOnce sync.Once handler Handler mu sync.Mutex recvBuf []byte } // Dial 连接服务器并发送握手。 func Dial(addr string, hs netproto.Handshake, h Handler) (*Client, error) { conn, err := net.DialTimeout("tcp", addr, 5*time.Second) if err != nil { return nil, fmt.Errorf("netclient.Dial %s: %w", addr, err) } c := &Client{ conn: conn, send: make(chan []byte, 64), done: make(chan struct{}), handler: h, } if _, err := conn.Write(netproto.EncodeFrame(netproto.Frame{MsgID: netproto.MsgHandshake, Payload: hs.Encode()})); err != nil { _ = conn.Close() return nil, fmt.Errorf("netclient.Dial 握手: %w", err) } go c.readLoop() go c.writeLoop() return c, nil } // readLoop 读循环:解帧分发(半包等待,协议错误断开)。 func (c *Client) readLoop() { tmp := make([]byte, 64*1024) for { _ = c.conn.SetReadDeadline(time.Now().Add(ioTimeout)) n, err := c.conn.Read(tmp) if err != nil { c.Close() return } c.mu.Lock() c.recvBuf = append(c.recvBuf, tmp[:n]...) for len(c.recvBuf) >= 4 { f, rest, err := netproto.DecodeFrame(c.recvBuf) if err != nil { break // 半包:等更多数据 } c.recvBuf = rest c.dispatch(f) } c.mu.Unlock() } } // dispatch 消息分发。 func (c *Client) dispatch(f netproto.Frame) { switch f.MsgID { case netproto.MsgLoginResponse: var code uint8 if len(f.Payload) > 0 { code = f.Payload[0] } c.handler.OnLogin(code) case netproto.MsgChunkData: c.handler.OnChunk(f.Payload) case netproto.MsgBlockChange: if m, err := netproto.DecodeBlockChange(f.Payload); err == nil { c.handler.OnBlockChange(m) } case netproto.MsgEntityMove: if m, err := netproto.DecodeEntityMove(f.Payload); err == nil { c.handler.OnEntityMove(m) } case netproto.MsgDisconnect: c.Close() } } // writeLoop 写循环。 func (c *Client) writeLoop() { for { select { case <-c.done: return case data := <-c.send: _ = c.conn.SetWriteDeadline(time.Now().Add(ioTimeout)) if _, err := c.conn.Write(data); err != nil { c.Close() return } } } } // SendMove 发送玩家移动(20Hz,多人同步.md §3)。 func (c *Client) SendMove(m netproto.PlayerMove) { c.sendFrame(netproto.Frame{MsgID: netproto.MsgPlayerMove, Payload: m.Encode()}) } // SendHeartbeat 发送心跳(网络协议.md §7)。 func (c *Client) SendHeartbeat(seq uint8) { c.sendFrame(netproto.Frame{MsgID: netproto.MsgHeartbeat, Payload: []byte{seq}}) } // sendFrame 入队发送(非阻塞,队列满丢弃)。 func (c *Client) sendFrame(f netproto.Frame) { if c.closed.Load() { return } select { case c.send <- netproto.EncodeFrame(f): default: } } // Close 关闭连接(幂等)。 func (c *Client) Close() { c.closeOnce.Do(func() { c.closed.Store(true) close(c.done) _ = c.conn.Close() }) } // Closed 是否已关闭。 func (c *Client) Closed() bool { return c.closed.Load() }