diff --git a/actor/mailbox.go b/actor/mailbox.go new file mode 100644 index 000000000..4d15f434c --- /dev/null +++ b/actor/mailbox.go @@ -0,0 +1,176 @@ +package actor + +import ( + "context" + "iter" + "sync" + "sync/atomic" +) + +// Mailbox represents the message queue for an actor. It provides methods for +// sending messages and receiving them via an iterator pattern. +type Mailbox[M Message, R any] interface { + // Send attempts to send an envelope to the mailbox with context-based + // cancellation. Returns true if sent successfully, false if the + // context was cancelled or the mailbox is closed. + Send(ctx context.Context, env envelope[M, R]) bool + + // TrySend attempts to send without blocking. Returns true if the + // envelope was sent, false if the mailbox is full or closed. + TrySend(env envelope[M, R]) bool + + // Receive returns an iterator for consuming messages from the mailbox. + // The iterator will yield messages until the mailbox is closed or the + // context is cancelled. + Receive(ctx context.Context) iter.Seq[envelope[M, R]] + + // Close closes the mailbox, preventing new messages from being sent. + // Any remaining messages can still be consumed via Receive. + Close() + + // IsClosed returns true if the mailbox has been closed. + IsClosed() bool + + // Drain returns an iterator that yields all remaining messages in the + // mailbox after it has been closed. This is useful for cleanup. + Drain() iter.Seq[envelope[M, R]] +} + +// ChannelMailbox is a channel-based implementation of the Mailbox interface. +type ChannelMailbox[M Message, R any] struct { + ch chan envelope[M, R] + closed atomic.Bool + + // mu protects Send/TrySend operations to prevent send-on-closed-channel + // panics. Close() acquires write lock, Send/TrySend acquire read lock. + mu sync.RWMutex + + // closeOnce ensures Close() executes exactly once. + closeOnce sync.Once + + // actorCtx is the actor's context for lifecycle management. + actorCtx context.Context +} + +// NewChannelMailbox creates a new channel-based mailbox with the specified +// buffer capacity and actor context. +func NewChannelMailbox[M Message, R any](actorCtx context.Context, + capacity int) *ChannelMailbox[M, R] { + + if capacity <= 0 { + capacity = 1 + } + return &ChannelMailbox[M, R]{ + ch: make(chan envelope[M, R], capacity), + actorCtx: actorCtx, + } +} + +// Send implements Mailbox.Send with context-aware blocking send. +func (m *ChannelMailbox[M, R]) Send(ctx context.Context, + env envelope[M, R]) bool { + + m.mu.RLock() + defer m.mu.RUnlock() + + if m.IsClosed() { + return false + } + + select { + case m.ch <- env: + return true + case <-ctx.Done(): + return false + case <-m.actorCtx.Done(): + // Actor is shutting down. + return false + } +} + +// TrySend implements Mailbox.TrySend with non-blocking send. +func (m *ChannelMailbox[M, R]) TrySend(env envelope[M, R]) bool { + m.mu.RLock() + defer m.mu.RUnlock() + + if m.IsClosed() { + return false + } + + select { + case m.ch <- env: + return true + default: + return false + } +} + +// Receive implements Mailbox.Receive using iter.Seq pattern. +func (m *ChannelMailbox[M, R]) Receive( + ctx context.Context) iter.Seq[envelope[M, R]] { + return func(yield func(envelope[M, R]) bool) { + for { + select { + case env, ok := <-m.ch: + if !ok { + return + } + + if !yield(env) { + return + } + + case <-ctx.Done(): + return + + case <-m.actorCtx.Done(): + return + } + } + } +} + +// Close implements Mailbox.Close. +func (m *ChannelMailbox[M, R]) Close() { + m.closeOnce.Do(func() { + m.mu.Lock() + defer m.mu.Unlock() + + m.closed.Store(true) + + close(m.ch) + }) +} + +// IsClosed implements Mailbox.IsClosed. +func (m *ChannelMailbox[M, R]) IsClosed() bool { + return m.closed.Load() +} + +// Drain implements Mailbox.Drain for cleanup after close. +func (m *ChannelMailbox[M, R]) Drain() iter.Seq[envelope[M, R]] { + return func(yield func(envelope[M, R]) bool) { + // Only drain if closed. + if !m.IsClosed() { + return + } + + // Drain all remaining messages from the channel. + for { + select { + case env, ok := <-m.ch: + // Channel closed, nothing left to drain. + if !ok { + return + } + + if !yield(env) { + return + } + default: + // Channel empty, done draining. + return + } + } + } +}