《MMO 服务器开发》4.1 连接位置与路由目录

固定头 11 字节：length(4, BE) | cmd(2, BE) | seq(4, BE) | flags(1)
length = 7 + N（7 = cmd(2)+seq(4)+flags(1)），帧总长 = 11 + N
HeaderSize = 11，MaxPayload = 1 << 20

// file: internal/gateway/directory.go
// tier: L1
// verify: go test ./internal/gateway/
package gateway

import "sync"

// PlayerID 是全服唯一角色 ID（卷三共享类型）。
type PlayerID uint64

// NodeID 是集群节点标识，形如 gateway-03。
type NodeID string

// Binding 是「某个玩家当前挂在哪个网关」的一条路由记录。Gen 是绑定代次：
// 同一玩家换服时，旧连接迟到的断开事件带的是旧代次，只有代次吻合才允许
// 摘除，否则会把新连接刚建立的绑定误删。
type Binding struct {
	Node NodeID
	Gen  uint64
}

// RouteDirectory 维护「玩家 → 网关节点」的路由目录。它是整个网关集群里唯一
// 知道「谁在哪」的地方：客户端重连、场景服回推、跨服调用都查它。
type RouteDirectory struct {
	mu     sync.RWMutex
	routes map[PlayerID]Binding
	seq    uint64
}

// NewRouteDirectory 返回一个空目录。
func NewRouteDirectory() *RouteDirectory {
	return &RouteDirectory{routes: make(map[PlayerID]Binding)}
}

// Bind 把玩家绑定到某网关节点，返回本次绑定的代次。若该玩家已挂在别的
// 节点上，旧绑定被覆盖——这正是「换服」：调用方拿到新代次后，负责去关
// 旧连接；旧连接后续的断开事件会因为代次不匹配而被忽略。
func (d *RouteDirectory) Bind(p PlayerID, node NodeID) uint64 {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.seq++
	d.routes[p] = Binding{Node: node, Gen: d.seq}
	return d.seq
}

// Lookup 查玩家当前所在节点。
func (d *RouteDirectory) Lookup(p PlayerID) (NodeID, bool) {
	d.mu.RLock()
	defer d.mu.RUnlock()
	b, ok := d.routes[p]
	return b.Node, ok
}

// LookupGen 同时返回代次，供网关在排空时判断自己是否仍持有该玩家。
func (d *RouteDirectory) LookupGen(p PlayerID) (Binding, bool) {
	d.mu.RLock()
	defer d.mu.RUnlock()
	b, ok := d.routes[p]
	return b, ok
}

// Unbind 摘除绑定，但只在该代次仍是最新时生效。旧连接的迟到断开事件带的
// 是旧代次，比对失败即被忽略，不会误删新连接建立的绑定。
func (d *RouteDirectory) Unbind(p PlayerID, gen uint64) bool {
	d.mu.Lock()
	defer d.mu.Unlock()
	b, ok := d.routes[p]
	if !ok || b.Gen != gen {
		return false
	}
	delete(d.routes, p)
	return true
}

// Count 返回在线绑定数。
func (d *RouteDirectory) Count() int {
	d.mu.RLock()
	defer d.mu.RUnlock()
	return len(d.routes)
}

// NodeLoad 统计每个节点当前持有的连接数，排空时用它判断某节点是否已经空了。
func (d *RouteDirectory) NodeLoad() map[NodeID]int {
	d.mu.RLock()
	defer d.mu.RUnlock()
	out := make(map[NodeID]int)
	for _, b := range d.routes {
		out[b.Node]++
	}
	return out
}

// snippet: internal/gateway/directory_test.go 节选
// TestStaleUnbindIgnored 是本节的核心用例：玩家换服时，旧连接迟到的断开
// 事件不能把新连接刚建立的绑定删掉。代次不匹配即拒绝摘除。
func TestStaleUnbindIgnored(t *testing.T) {
	d := NewRouteDirectory()
	oldGen := d.Bind(2002, "gateway-01") // 玩家先挂在 gw-01
	newGen := d.Bind(2002, "gateway-02") // 换服到 gw-02，旧绑定被覆盖

	if d.Unbind(2002, oldGen) { // 旧连接迟到的断开事件
		t.Fatal("stale unbind must be rejected")
	}
	if node, ok := d.Lookup(2002); !ok || node != "gateway-02" {
		t.Fatalf("after stale unbind lookup = %q,%v want gateway-02,true", node, ok)
	}
	if !d.Unbind(2002, newGen) { // 新连接正常断开
		t.Fatal("current-gen unbind should succeed")
	}
}

// file: cmd/gw/main.go
// tier: L2
// verify: go build ./cmd/gw/ && go test ./internal/gateway/ -run TestL2RouteDirectory
package main

import (
	"flag"
	"log"
	"net"
	"sync"

	"aegis/internal/gateway"
)

// gw 是一个最小网关节点：启动时向路由目录注册自己，之后每接一个客户端，
// 就把「玩家 → 本节点」写进目录；连接断开时按代次解绑。
type gw struct {
	node string
	dir  *dirConn
}

// dirConn 是到目录的一条长连接。目录操作是请求/应答式，用互斥量串行化，
// 保证一次只有一条在途请求，读到的应答不会串台。
type dirConn struct {
	mu   sync.Mutex
	conn net.Conn
	seq  uint32
}

func dialDir(addr, node string) (*dirConn, error) {
	c, err := net.Dial("tcp", addr)
	if err != nil {
		return nil, err
	}
	d := &dirConn{conn: c}
	// 注册节点：CmdNodeRegisterReq 0x0F01，载荷是节点名。
	if err := gateway.WriteFrame(c, gateway.Frame{Cmd: gateway.CmdNodeRegisterReq, Payload: []byte(node)}); err != nil {
		return nil, err
	}
	if _, err := gateway.ReadFrame(c); err != nil { // 读注册应答
		return nil, err
	}
	return d, nil
}

func (d *dirConn) call(f gateway.Frame) (gateway.Frame, error) {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.seq++
	f.Seq = d.seq
	if err := gateway.WriteFrame(d.conn, f); err != nil {
		return gateway.Frame{}, err
	}
	return gateway.ReadFrame(d.conn)
}

func (d *dirConn) bind(p gateway.PlayerID, node string) (uint64, error) {
	r, err := d.call(gateway.Frame{Cmd: gateway.CmdRouteTableSync, Flags: gateway.OpBind, Payload: gateway.EncodeBind(p, gateway.NodeID(node))})
	if err != nil {
		return 0, err
	}
	return gateway.GetU64(r.Payload), nil
}

func (d *dirConn) unbind(p gateway.PlayerID, gen uint64) error {
	_, err := d.call(gateway.Frame{Cmd: gateway.CmdRouteTableSync, Flags: gateway.OpUnbind, Payload: gateway.EncodeUnbind(p, gen)})
	return err
}

// serve 处理一条客户端连接：登录帧 → 绑定 → 保持在线 → 断开时解绑。
func (g *gw) serve(c net.Conn) {
	defer c.Close()
	f, err := gateway.ReadFrame(c)
	if err != nil {
		return
	}
	if f.Cmd != 0x0010 || len(f.Payload) < 8 { // CmdLoginReq
		return
	}
	pid := gateway.PlayerID(gateway.GetU64(f.Payload[:8]))
	gen, err := g.dir.bind(pid, g.node)
	if err != nil {
		return
	}
	// 应答里带代次与节点名，客户端据此知道自己被路由到了哪个网关。
	ack := make([]byte, 8+len(g.node))
	gateway.PutU64(ack, gen)
	copy(ack[8:], g.node)
	if err := gateway.WriteFrame(c, gateway.Frame{Cmd: 0x0011, Seq: f.Seq, Payload: ack}); err != nil {
		return
	}
	// 保持在线：应答心跳，直到客户端关闭或出错。
	for {
		f, err := gateway.ReadFrame(c)
		if err != nil {
			break
		}
		if f.Cmd == 0x0001 { // CmdHeartbeatReq
			_ = gateway.WriteFrame(c, gateway.Frame{Cmd: 0x0002, Seq: f.Seq})
		}
	}
	_ = g.dir.unbind(pid, gen)
}

func main() {
	dirAddr := flag.String("dir", "127.0.0.1:32001", "route directory address")
	node := flag.String("node", "gateway-01", "node id")
	listen := flag.String("listen", "127.0.0.1:32011", "client listen address")
	flag.Parse()

	d, err := dialDir(*dirAddr, *node)
	if err != nil {
		log.Fatalf("gw %s: dial dir: %v", *node, err)
	}
	ln, err := net.Listen("tcp", *listen)
	if err != nil {
		log.Fatalf("gw %s: listen: %v", *node, err)
	}
	log.Printf("gw %s listening on %s", *node, *listen)
	g := &gw{node: *node, dir: d}
	for {
		c, err := ln.Accept()
		if err != nil {
			return
		}
		go g.serve(c)
	}
}