《游戏服务端编程实践》11.2 RPC 框架与消息总线

// file: ch11/rpcbus/rpcbus.go
// tier: L1
// verify: go test ./ch11/rpcbus/

package rpcbus

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

var (
	ErrEmptyMethod     = errors.New("rpcbus: empty method")
	ErrDuplicateMethod = errors.New("rpcbus: duplicate method")
	ErrEmptyTopic      = errors.New("rpcbus: empty topic")
)

// ErrorKind 区分传输层与业务层错误，重试策略只能建立在稳定错误码上。
type ErrorKind string

const (
	Transport ErrorKind = "transport"
	Business  ErrorKind = "business"
)

type RPCError struct {
	Kind    ErrorKind
	Code    string
	Message string
}

func (e *RPCError) Error() string {
	return fmt.Sprintf("rpcbus: %s error %s: %s", e.Kind, e.Code, e.Message)
}

func NewTransportError(code, message string) *RPCError {
	return &RPCError{Kind: Transport, Code: code, Message: message}
}

func NewBusinessError(code, message string) *RPCError {
	return &RPCError{Kind: Business, Code: code, Message: message}
}

// CodeOf 提取稳定错误码；日志、告警和重试都不应依赖错误文本。
func CodeOf(err error) string {
	var rpcErr *RPCError
	if errors.As(err, &rpcErr) {
		return rpcErr.Code
	}
	return ""
}

type Handler func(ctx context.Context, payload []byte) ([]byte, error)

// Server 保存方法表；handler 必须自己检查 ctx，客户端超时只是最后一道保险。
type Server struct {
	mu       sync.RWMutex
	handlers map[string]Handler
}

func NewServer() *Server {
	return &Server{handlers: make(map[string]Handler)}
}

func (s *Server) Register(method string, handler Handler) error {
	method = strings.TrimSpace(method)
	if method == "" {
		return ErrEmptyMethod
	}
	if handler == nil {
		return fmt.Errorf("rpcbus: nil handler for %s", method)
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if _, ok := s.handlers[method]; ok {
		return fmt.Errorf("%w: %s", ErrDuplicateMethod, method)
	}
	s.handlers[method] = handler
	return nil
}

func (s *Server) invoke(ctx context.Context, method string, payload []byte) ([]byte, error) {
	s.mu.RLock()
	handler, ok := s.handlers[method]
	s.mu.RUnlock()
	if !ok {
		return nil, NewTransportError("method_not_found", method)
	}
	return handler(ctx, payload)
}

// Client 统一施加超时，并把 handler 返回的错误规整成 RPCError。
type Client struct {
	server  *Server
	timeout time.Duration
}

func NewClient(server *Server, timeout time.Duration) *Client {
	return &Client{server: server, timeout: timeout}
}

func (c *Client) Call(ctx context.Context, method string, payload []byte) ([]byte, error) {
	if c.timeout > 0 {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, c.timeout)
		defer cancel()
	}
	type reply struct {
		payload []byte
		err     error
	}
	done := make(chan reply, 1)
	go func() {
		p, err := c.server.invoke(ctx, method, payload)
		done <- reply{payload: p, err: err}
	}()
	select {
	case r := <-done:
		return r.payload, normalizeError(r.err)
	case <-ctx.Done():
		return nil, NewTransportError("deadline_exceeded", ctx.Err().Error())
	}
}

func normalizeError(err error) error {
	if err == nil {
		return nil
	}
	var rpcErr *RPCError
	if errors.As(err, &rpcErr) {
		return rpcErr
	}
	if errors.Is(err, context.DeadlineExceeded) {
		return NewTransportError("deadline_exceeded", err.Error())
	}
	if errors.Is(err, context.Canceled) {
		return NewTransportError("canceled", err.Error())
	}
	return NewTransportError("internal", err.Error())
}

// Message 是总线投递单元；ID 由生产者给出，重复投递时消费者靠它去重。
type Message struct {
	ID      string
	Topic   string
	Payload []byte
	Seq     uint64
}

// Delivery 记录一次发布结果；Dropped 非零说明有慢消费者，需要重新全量对齐。
type Delivery struct {
	Seq       uint64
	Delivered int
	Dropped   int
}

type Subscription struct {
	id        uint64
	topic     string
	ch        chan Message
	bus       *Bus
	closeOnce sync.Once
}

func (s *Subscription) C() <-chan Message {
	return s.ch
}

func (s *Subscription) Close() {
	s.closeOnce.Do(func() { s.bus.unsubscribe(s) })
}

// Bus 提供至少一次投递；发布成功不等于消费成功，消费端必须按 ID 去重。
type Bus struct {
	mu      sync.RWMutex
	nextSub uint64
	seq     atomic.Uint64
	subs    map[string]map[uint64]*Subscription
}

func NewBus() *Bus {
	return &Bus{subs: make(map[string]map[uint64]*Subscription)}
}

func (b *Bus) Subscribe(topic string, buffer int) (*Subscription, error) {
	topic = strings.TrimSpace(topic)
	if topic == "" {
		return nil, ErrEmptyTopic
	}
	if buffer < 0 {
		buffer = 0
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	b.nextSub++
	sub := &Subscription{id: b.nextSub, topic: topic, ch: make(chan Message, buffer), bus: b}
	if b.subs[topic] == nil {
		b.subs[topic] = make(map[uint64]*Subscription)
	}
	b.subs[topic][sub.id] = sub
	return sub, nil
}

func (b *Bus) Publish(msg Message) (Delivery, error) {
	topic := strings.TrimSpace(msg.Topic)
	if topic == "" {
		return Delivery{}, ErrEmptyTopic
	}
	b.mu.RLock()
	defer b.mu.RUnlock()
	msg.Topic = topic
	msg.Seq = b.seq.Add(1)
	delivery := Delivery{Seq: msg.Seq}
	for _, sub := range b.subs[topic] {
		select {
		case sub.ch <- msg:
			delivery.Delivered++
		default:
			delivery.Dropped++
		}
	}
	return delivery, nil
}

func (b *Bus) unsubscribe(sub *Subscription) {
	b.mu.Lock()
	defer b.mu.Unlock()
	group, ok := b.subs[sub.topic]
	if !ok {
		return
	}
	if _, ok := group[sub.id]; !ok {
		return
	}
	delete(group, sub.id)
	if len(group) == 0 {
		delete(b.subs, sub.topic)
	}
	close(sub.ch)
}

cd .local/checks/gs-practice-lab && gofmt -l ch11/rpcbus && go vet ./ch11/rpcbus/ && go test ./ch11/rpcbus/ -count=1

ok  	gspractice/ch11/rpcbus	0.480s