《MMO 服务器开发》10.3 对照：Go channel 邮箱与 Erlang/Skynet 式 Actor

变体 A：有界 channel 邮箱              变体 B：无界队列 + 单消费者

  生产者 ─Send─▶ [ M M M ] ─▶ 消费者     生产者 ─Send─▶ [ M M M M M ... ] ─▶ 消费者
                   ▲ 满则阻塞                             ▲ 永不阻塞，可无限增长
                   └ 背压自动回传                          └ 背压必须由业务自己做

// file: internal/actor/actor.go
// tier: L1
// verify: go test ./internal/actor/ -run 'TestSelectiveReceiveIsQueueOnly|TestUnboundedQueueGrowsUnderSlowConsumer|TestBothVariantsApplySameCount'

package actor

import "sync"

// ChannelActor is variant A: the mailbox *is* a bounded Go channel and the
// consumer ranges over it. This is the idiomatic Go actor: blocking Send gives
// backpressure for free, but there is no selective receive and no way to peek.
type ChannelActor struct {
	ch chan Msg
}

// NewChannelActor returns a channel actor with the given mailbox capacity.
func NewChannelActor(capacity int) *ChannelActor {
	if capacity <= 0 {
		capacity = 1
	}
	return &ChannelActor{ch: make(chan Msg, capacity)}
}

// Send blocks when the mailbox is full.
func (a *ChannelActor) Send(msg Msg) { a.ch <- msg }

// Close stops the consumer after the queue drains.
func (a *ChannelActor) Close() { close(a.ch) }

// Run consumes until Close, applying every message in arrival order. A message
// cannot be skipped: Go channels have no selective receive.
func (a *ChannelActor) Run(apply func(Msg)) {
	for msg := range a.ch {
		apply(msg)
	}
}

// QueueActor is variant B: an explicit FIFO slice guarded by a mutex, with a
// condition variable waking a single consumer. It mirrors the Erlang/Skynet
// mailbox more closely — the queue can grow without a bound, and a consumer may
// scan it for a matching message instead of taking the head.
type QueueActor struct {
	mu     sync.Mutex
	cond   *sync.Cond
	q      []Msg
	closed bool
	peak   int
}

// NewQueueActor returns an unbounded queue actor.
func NewQueueActor() *QueueActor {
	a := &QueueActor{}
	a.cond = sync.NewCond(&a.mu)
	return a
}

// Send appends to the queue and wakes the consumer. It never blocks, which is
// exactly why an unbounded mailbox needs some other backpressure signal.
func (a *QueueActor) Send(msg Msg) {
	a.mu.Lock()
	a.q = append(a.q, msg)
	if len(a.q) > a.peak {
		a.peak = len(a.q)
	}
	a.mu.Unlock()
	a.cond.Signal()
}

// Close stops the consumer after the queue drains.
func (a *QueueActor) Close() {
	a.mu.Lock()
	a.closed = true
	a.mu.Unlock()
	a.cond.Broadcast()
}

// Peak returns the high-water mark of the queue. An unbounded mailbox has no
// cap, so this is the only backpressure signal an operator ever sees.
func (a *QueueActor) Peak() int {
	a.mu.Lock()
	defer a.mu.Unlock()
	return a.peak
}

func (a *QueueActor) pop() (Msg, bool) {
	a.mu.Lock()
	defer a.mu.Unlock()
	for len(a.q) == 0 && !a.closed {
		a.cond.Wait()
	}
	if len(a.q) == 0 {
		return Msg{}, false
	}
	msg := a.q[0]
	a.q = a.q[1:]
	return msg, true
}

// Run consumes until Close, applying every message in arrival order.
func (a *QueueActor) Run(apply func(Msg)) {
	for {
		msg, ok := a.pop()
		if !ok {
			return
		}
		apply(msg)
	}
}

// ReceiveFirst scans the queue for the first message matching pred, leaving the
// rest in place. This is Erlang-style selective receive, and it is exactly what
// a Go channel cannot do.
func (a *QueueActor) ReceiveFirst(pred func(Msg) bool) (Msg, bool) {
	a.mu.Lock()
	defer a.mu.Unlock()
	for i, msg := range a.q {
		if pred(msg) {
			a.q = append(a.q[:i], a.q[i+1:]...)
			return msg, true
		}
	}
	return Msg{}, false
}

// snippet: internal/actor/actor_test.go（节选）

func TestSelectiveReceiveIsQueueOnly(t *testing.T) {
	qa := NewQueueActor()
	qa.Send(Msg{Cmd: CmdMoveReq, Seq: 1})
	qa.Send(Msg{Cmd: CmdChatSend, Seq: 2})
	qa.Send(Msg{Cmd: CmdMoveReq, Seq: 3})

	got, ok := qa.ReceiveFirst(func(m Msg) bool { return m.Cmd == CmdChatSend })
	if !ok || got.Seq != 2 {
		t.Fatalf("selective receive = %+v ok=%v, want seq 2", got, ok)
	}

	// The two moves are still queued, in order.
	first, _ := qa.ReceiveFirst(func(m Msg) bool { return m.Cmd == CmdMoveReq })
	second, _ := qa.ReceiveFirst(func(m Msg) bool { return m.Cmd == CmdMoveReq })
	if first.Seq != 1 || second.Seq != 3 {
		t.Fatalf("leftovers = %d,%d want 1,3", first.Seq, second.Seq)
	}
}

// snippet: internal/actor/actor_test.go（节选）

func TestUnboundedQueueGrowsUnderSlowConsumer(t *testing.T) {
	qa := NewQueueActor()
	const n = 20000
	for i := 0; i < n; i++ {
		qa.Send(Msg{Cmd: CmdMoveReq, Seq: uint32(i)})
	}
	// Nothing consumed yet: the high-water mark is the whole burst.
	if got := qa.Peak(); got != n {
		t.Fatalf("peak = %d, want %d", got, n)
	}
}