《MMO 服务器开发》6.2 跨服聊天与消息路由

// snippet: NATS subject 约定（测试前缀 aegis.test.ch06.）
aegis.test.ch06.chan.<channel>   —— 频道广播：谁订阅谁收，扇出
aegis.test.ch06.srv.<serverID>   —— 服收件箱：该服在线玩家的定向消息都投这里
// 频道消息与定向消息共用一个 ChatMessage 载荷，靠 Target 是否非 0 区分。

                  ┌──────────────── aegis.test.ch06.* ──────────────────┐
                  │                                                    │
   服 301 ─sub──► │  chan.world        ◄─pub─ 任意服的聊天入口（扇出）  │
   服 303 ─sub──► │  （302 没订阅 → 收不到世界频道）                     │
                  │                                                    │
   服 301 ─sub──► │  srv.301   ◄─pub─ SendTo(5001) 查目录 → 在 301       │
   服 303 ─sub──► │  srv.303   ◄─pub─ SendTo(5002) 查目录 → 在 303       │
                  │                                                    │
                  └────────────────────────────────────────────────────┘

// file: internal/crossserver/router.go
// tier: L2
// verify: go test ./internal/crossserver/ -run TestCrossServerChatRouting
package crossserver

import (
	"strconv"
	"sync"

	"github.com/nats-io/nats.go"
)

// Router 用 NATS 做跨服聊天：频道消息按「订阅关系」扇出，定向消息按玩家路由。
//
//	<prefix>chan.<channel>  —— 频道广播：谁订阅了这个频道谁收
//	<prefix>srv.<serverID>  —— 服收件箱：该服在线玩家的定向消息都投这里
type Router struct {
	nc     *nats.Conn
	prefix string

	mu       sync.Mutex
	presence map[uint64]uint16 // PlayerID -> 当前所在服
	offline  map[uint64][]ChatMessage
}

func NewRouter(url, prefix string) (*Router, error) {
	nc, err := nats.Connect(url)
	if err != nil {
		return nil, err
	}
	return &Router{nc: nc, prefix: prefix, presence: make(map[uint64]uint16), offline: make(map[uint64][]ChatMessage)}, nil
}

func (r *Router) Close() { r.nc.Close() }

func (r *Router) chanSubject(channel string) string { return r.prefix + "chan." + channel }
func (r *Router) srvSubject(serverID uint16) string { return r.prefix + "srv." + strconv.Itoa(int(serverID)) }

// SubscribeChannel 让某个服订阅一个跨服频道。
func (r *Router) SubscribeChannel(channel string, fn func(ChatMessage)) (*nats.Subscription, error) {
	return r.nc.Subscribe(r.chanSubject(channel), func(m *nats.Msg) {
		if c, err := DecodeChatMessage(m.Data); err == nil {
			fn(c)
		}
	})
}

// SubscribeServer 订阅一个服的收件箱（定向消息的落点）。
func (r *Router) SubscribeServer(serverID uint16, fn func(ChatMessage)) (*nats.Subscription, error) {
	return r.nc.Subscribe(r.srvSubject(serverID), func(m *nats.Msg) {
		if c, err := DecodeChatMessage(m.Data); err == nil {
			fn(c)
		}
	})
}

// PublishChannel 往频道广播：扇出给所有订阅了该频道的服。
func (r *Router) PublishChannel(channel string, msg ChatMessage) error {
	msg.Channel = channel
	return r.nc.Publish(r.chanSubject(channel), msg.Encode())
}

// SendTo 把定向消息投给某个玩家：在线则投其所在服的收件箱，离线则暂存。
func (r *Router) SendTo(pid uint64, msg ChatMessage) error {
	r.mu.Lock()
	sid, online := r.presence[pid]
	if !online {
		r.offline[pid] = append(r.offline[pid], msg)
	}
	r.mu.Unlock()
	if !online {
		return nil
	}
	msg.Target = pid
	return r.nc.Publish(r.srvSubject(sid), msg.Encode())
}

// OnLogin 玩家上线：登记所在服，并把断线期间攒下的消息补投到新服的收件箱。
func (r *Router) OnLogin(pid uint64, serverID uint16) (int, error) {
	r.mu.Lock()
	r.presence[pid] = serverID
	pending := r.offline[pid]
	delete(r.offline, pid)
	r.mu.Unlock()
	for _, m := range pending {
		m.Target = pid
		if err := r.nc.Publish(r.srvSubject(serverID), m.Encode()); err != nil {
			return 0, err
		}
	}
	return len(pending), nil
}

// OnLogout 玩家下线：清掉在线登记，后续定向消息转入离线暂存。
func (r *Router) OnLogout(pid uint64) {
	r.mu.Lock()
	delete(r.presence, pid)
	r.mu.Unlock()
}

// Pending 返回某玩家离线暂存的消息数。
func (r *Router) Pending(pid uint64) int {
	r.mu.Lock()
	defer r.mu.Unlock()
	return len(r.offline[pid])
}

// Flush 等 NATS 把已发布的消息处理完。
func (r *Router) Flush() error { return r.nc.Flush() }

// snippet: internal/crossserver/chat_test.go（断言片段）
// 1) 世界频道扇出：只到订阅了 world 的两个服，且不落进任何收件箱。
if !waitFor(t, time.Second, func() bool { return worldSinks[301].n() == 1 && worldSinks[303].n() == 1 }) {
	t.Fatalf("world 扇出失败: 301=%d 303=%d", worldSinks[301].n(), worldSinks[303].n())
}
for id, s := range inbox {
	if s.n() != 0 {
		t.Fatalf("频道广播误入了服 %d 的收件箱（%d 条）", id, s.n())
	}
}
// 2) 定向路由：5001 在 301、5002 在 303，各自只进对应收件箱。
if err := r.SendTo(5001, ChatMessage{From: 900, Seq: 2, Body: "私聊 5001"}); err != nil {
	t.Fatalf("SendTo: %v", err)
}
if !waitFor(t, time.Second, func() bool { return inbox[301].n() == 1 && inbox[303].n() == 1 }) {
	t.Fatalf("定向路由失败: 301=%d 303=%d", inbox[301].n(), inbox[303].n())
}
if inbox[302].n() != 0 {
	t.Fatalf("服 302 不该收到任何定向消息，实收 %d", inbox[302].n())
}

SendTo(6001)  ─► presence 无 6001 ─► 进 offline[6001]  ─► 收件箱 0 条
SendTo(6001)  ─► presence 无 6001 ─► 进 offline[6001]  ─► 收件箱 0 条，Pending=2
OnLogin(6001,303) ─► presence[6001]=303 ─► 逐条 pub srv.303 ─► 收件箱 2 条，Pending=0
OnLogout(6001)    ─► 删 presence[6001]
SendTo(6001)  ─► presence 无 6001 ─► 又进 offline[6001] ─► Pending=1

=== RUN   TestCrossServerChatRouting
--- PASS: TestCrossServerChatRouting (0.03s)
=== RUN   TestOfflineBufferThenFlush
--- PASS: TestOfflineBufferThenFlush (0.12s)
=== RUN   TestChatMessageRoundTrip
--- PASS: TestChatMessageRoundTrip (0.00s)
ok  	aegis/internal/crossserver	1.692s