《游戏服务端编程实践》5.4.2 Redis / gRPC / Akka 通信实现

// 伪代码：广播踢下线，不用于关键业务
PUBLISH gw:kick  {"player_id":10086,"reason":"duplicate_login"}
SUBSCRIBE gw:kick
// 收到消息后：校验 player_id 是否在本网关，是则关闭对应 Session

// 伪代码：房间 Actor 的收件箱策略
room = actor("room-1001")
room ! Join(player=42)          // 入房，串行处理
room ! Move(player=42, x=10)    // 与 Join 保序
// mailbox: capacity=1024, overflow=drop-new 并回错误码

// file: ch05/backend/rpcbackend.go
// tier: L1
// verify: go test ./ch05/backend/

package backend

import (
	"bytes"
	"context"
	"errors"
	"net"
	"net/rpc"
	"sync"
	"time"
)

type ForwardArgs struct{ Frame []byte }
type ForwardReply struct{ Frame []byte }

// RPCService 运行在后端进程，把整帧字节交给业务处理函数，不拆 Body。
type RPCService struct{ Handle func([]byte) ([]byte, error) }

func (s *RPCService) Forward(args ForwardArgs, reply *ForwardReply) error {
	if s.Handle == nil {
		return errors.New("rpcbackend: no handler")
	}
	out, err := s.Handle(args.Frame)
	if err != nil {
		return err
	}
	reply.Frame = out
	return nil
}

// ServeRPC 在 listener 上注册服务，每个连接一个 goroutine。
func ServeRPC(ln net.Listener, svc *RPCService) error {
	srv := rpc.NewServer()
	if err := srv.RegisterName("Gateway", svc); err != nil {
		return err
	}
	for {
		conn, err := ln.Accept()
		if err != nil {
			return err
		}
		go srv.ServeConn(conn)
	}
}

// RPCBackend 复用一条 net/rpc 连接。超时或调用失败时关闭连接，
// 否则迟到的响应会被下一条请求读到。
type RPCBackend struct {
	mu      sync.Mutex
	addr    string
	timeout time.Duration
	client  *rpc.Client
}

func NewRPCBackend(addr string, timeout time.Duration) *RPCBackend {
	return &RPCBackend{addr: addr, timeout: timeout}
}

func (b *RPCBackend) connLocked() (*rpc.Client, error) {
	if b.client != nil {
		return b.client, nil
	}
	client, err := rpc.Dial("tcp", b.addr)
	if err != nil {
		return nil, err
	}
	b.client = client
	return client, nil
}

func (b *RPCBackend) dropLocked() {
	if b.client != nil {
		_ = b.client.Close()
		b.client = nil
	}
}

func (b *RPCBackend) RoundTrip(ctx context.Context, req Frame) (Frame, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if err := ctx.Err(); err != nil {
		return Frame{}, err
	}
	client, err := b.connLocked()
	if err != nil {
		return Frame{}, err
	}
	call := client.Go("Gateway.Forward", ForwardArgs{Frame: req.Encode()}, &ForwardReply{}, make(chan *rpc.Call, 1))
	timeout := b.timeout
	if deadline, ok := ctx.Deadline(); ok {
		if left := time.Until(deadline); left < timeout {
			timeout = left
		}
	}
	timer := time.NewTimer(timeout)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		b.dropLocked()
		return Frame{}, ctx.Err()
	case <-timer.C:
		// net/rpc 不能取消在途调用；断开连接丢弃迟到响应，防止串包。
		b.dropLocked()
		return Frame{}, context.DeadlineExceeded
	case res := <-call.Done:
		if res.Error != nil {
			b.dropLocked()
			return Frame{}, res.Error
		}
		resp, err := ReadFrame(bytes.NewReader(res.Reply.(*ForwardReply).Frame), maxFrameSize)
		if err != nil {
			b.dropLocked()
			return Frame{}, err
		}
		resp.RequestID, resp.Route, resp.Type = req.RequestID, req.Route, FrameResponse
		return resp, nil
	}
}

func (b *RPCBackend) Close() error {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.dropLocked()
	return nil
}

// file: ch05/backend/rpcbackend_test.go
// tier: L1
// verify: go test ./ch05/backend/

package backend

import (
	"bytes"
	"context"
	"errors"
	"net"
	"testing"
	"time"
)

func startRPC(t *testing.T, svc *RPCService) string {
	t.Helper()
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	go func() { _ = ServeRPC(ln, svc) }()
	t.Cleanup(func() { _ = ln.Close() })
	return ln.Addr().String()
}

func TestRPCBackendRoundTrip(t *testing.T) {
	addr := startRPC(t, &RPCService{Handle: func(raw []byte) ([]byte, error) {
		req, err := ReadFrame(bytes.NewReader(raw), maxFrameSize)
		if err != nil {
			return nil, err
		}
		if req.RequestID != 99 || !bytes.Equal(req.Body, []byte("payload")) {
			return nil, errors.New("rpcbackend: altered frame")
		}
		return Frame{Type: FrameResponse, Route: req.Route, Body: append([]byte("echo:"), req.Body...)}.Encode(), nil
	}})
	b := NewRPCBackend(addr, time.Second)
	defer b.Close()
	resp, err := b.RoundTrip(context.Background(), Frame{Type: FrameRequest, Route: 7, RequestID: 99, Body: []byte("payload")})
	if err != nil {
		t.Fatal(err)
	}
	if resp.RequestID != 99 || resp.Route != 7 || resp.Type != FrameResponse || string(resp.Body) != "echo:payload" {
		t.Fatalf("resp = %+v", resp)
	}
}

func TestRPCBackendTimeout(t *testing.T) {
	addr := startRPC(t, &RPCService{Handle: func([]byte) ([]byte, error) {
		time.Sleep(200 * time.Millisecond)
		return nil, errors.New("rpcbackend: late")
	}})
	b := NewRPCBackend(addr, 30*time.Millisecond)
	defer b.Close()
	start := time.Now()
	_, err := b.RoundTrip(context.Background(), Frame{Type: FrameRequest, Route: 7, RequestID: 1})
	if !errors.Is(err, context.DeadlineExceeded) || time.Since(start) > 180*time.Millisecond {
		t.Fatalf("err = %v, elapsed = %v", err, time.Since(start))
	}
}

cd .local/checks/gs-practice-lab
gofmt -l ch05/backend
go vet ./ch05/backend/
go test ./ch05/backend/ -run 'RPC' -count=1 -v