actor: refactor Actor to use Mailbox interface
This commit refactors the Actor implementation to use the new Mailbox interface instead of directly managing a channel. This change significantly simplifies the actor's message processing loop and improves separation of concerns. The main changes include replacing the direct channel field with a Mailbox interface, updating NewActor to create a ChannelMailbox instance, and refactoring the process method to use the iterator pattern provided by mailbox.Receive. The new implementation uses a clean for-range loop over the mailbox's message iterator, eliminating the complex select statement that previously handled both message reception and context cancellation. The Tell and Ask methods in actorRefImpl have been simplified to use the mailbox's Send method, which internally handles both the caller's context and the actor's context. This eliminates the need for complex select statements in these methods and ensures consistent context handling throughout the actor system. Message draining during shutdown is now handled through the mailbox's Drain method, providing a cleaner separation between normal message processing and cleanup operations. The actor still properly sends unprocessed messages to the Dead Letter Office and completes pending promises with appropriate errors during shutdown.
This commit is contained in:
Olaoluwa Osuntokun committed
1 parent
de7b0e1dd4
commit
20c5c9d6be
1 file changed
+44
-97
+44
-97
@@ -45,7 +45,7 @@ type Actor[M Message, R any] struct {
|
||||
behavior ActorBehavior[M, R]
|
||||
|
||||
// mailbox is the incoming message queue for the actor.
|
||||
mailbox chan envelope[M, R]
|
||||
mailbox Mailbox[M, R]
|
||||
|
||||
// ctx is the context governing the actor's lifecycle.
|
||||
ctx context.Context
|
||||
@@ -93,10 +93,13 @@ func NewActor[M Message, R any](cfg ActorConfig[M, R]) (*Actor[M, R],
|
||||
mailboxCapacity = 1
|
||||
}
|
||||
|
||||
// Create mailbox - could be injected via config in the future.
|
||||
mailbox := NewChannelMailbox[M, R](ctx, mailboxCapacity)
|
||||
|
||||
actor := &Actor[M, R]{
|
||||
id: cfg.ID,
|
||||
behavior: cfg.Behavior,
|
||||
mailbox: make(chan envelope[M, R], mailboxCapacity),
|
||||
mailbox: mailbox,
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
dlo: cfg.DLO,
|
||||
@@ -123,76 +126,32 @@ func (a *Actor[M, R]) Start() {
|
||||
// process is the main event loop for the actor. It continuously monitors its
|
||||
// mailbox for incoming messages and its context for cancellation signals.
|
||||
func (a *Actor[M, R]) process() {
|
||||
for {
|
||||
select {
|
||||
case env := <-a.mailbox:
|
||||
result := a.behavior.Receive(a.ctx, env.message)
|
||||
// Use the new iterator pattern for receiving messages.
|
||||
for env := range a.mailbox.Receive(a.ctx) {
|
||||
result := a.behavior.Receive(a.ctx, env.message)
|
||||
|
||||
// If a promise was provided (i.e., it was an "ask"
|
||||
// operation), complete the promise with the result from
|
||||
// the behavior.
|
||||
if env.promise != nil {
|
||||
env.promise.Complete(result)
|
||||
}
|
||||
// If a promise was provided (i.e., it was an "ask"
|
||||
// operation), complete the promise with the result from
|
||||
// the behavior.
|
||||
if env.promise != nil {
|
||||
env.promise.Complete(result)
|
||||
}
|
||||
}
|
||||
|
||||
// The actor's context has been cancelled, signaling a stop
|
||||
// request. Exit the processing loop to terminate the actor's
|
||||
// goroutine. Before exiting, drain any remaining messages from
|
||||
// the mailbox.
|
||||
//
|
||||
// NOTE: We intentionally do NOT close the mailbox channel here.
|
||||
// Closing it would create a TOCTOU race with Tell/Ask, which
|
||||
// check ctx.Err() before sending to the mailbox. Between that
|
||||
// check and the actual send, the channel could be closed,
|
||||
// causing a panic. Instead, we drain using a non-blocking
|
||||
// select loop. Messages sent after this point will remain in
|
||||
// the buffered channel and be garbage collected.
|
||||
case <-a.ctx.Done():
|
||||
log.Debugf("Actor %s: context cancelled, draining "+
|
||||
"mailbox", a.id)
|
||||
// Context was cancelled or mailbox closed, drain remaining messages.
|
||||
a.mailbox.Close()
|
||||
|
||||
// Drain any remaining messages from the mailbox
|
||||
// without closing the channel.
|
||||
drained := 0
|
||||
for {
|
||||
select {
|
||||
case env := <-a.mailbox:
|
||||
drained++
|
||||
for env := range a.mailbox.Drain() {
|
||||
// If a DLO is configured, send the original message there
|
||||
// for auditing or potential manual reprocessing.
|
||||
if a.dlo != nil {
|
||||
a.dlo.Tell(context.Background(), env.message)
|
||||
}
|
||||
|
||||
// If a DLO is configured, send the
|
||||
// original message there for auditing
|
||||
// or potential manual reprocessing.
|
||||
if a.dlo != nil {
|
||||
a.dlo.Tell(
|
||||
context.Background(),
|
||||
env.message,
|
||||
)
|
||||
}
|
||||
|
||||
// If it was an Ask, complete the
|
||||
// promise with an error indicating the
|
||||
// actor terminated.
|
||||
if env.promise != nil {
|
||||
env.promise.Complete(
|
||||
fn.Err[R](
|
||||
ErrActorTerminated,
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
default:
|
||||
if drained > 0 {
|
||||
log.Debugf("Actor %s: "+
|
||||
"drained %d "+
|
||||
"message(s) during "+
|
||||
"shutdown",
|
||||
a.id, drained)
|
||||
}
|
||||
|
||||
// No more messages in the mailbox.
|
||||
return
|
||||
}
|
||||
}
|
||||
// If it was an Ask, complete the promise with an error
|
||||
// indicating the actor terminated.
|
||||
if env.promise != nil {
|
||||
env.promise.Complete(fn.Err[R](ErrActorTerminated))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -227,22 +186,15 @@ func (ref *actorRefImpl[M, R]) Tell(ctx context.Context, msg M) {
|
||||
return
|
||||
}
|
||||
|
||||
select {
|
||||
// Message successfully enqueued in the actor's mailbox.
|
||||
case ref.actor.mailbox <- envelope[M, R]{message: msg, promise: nil}:
|
||||
env := envelope[M, R]{message: msg, promise: nil}
|
||||
|
||||
// The context for the Tell operation was cancelled before the message
|
||||
// could be enqueued. The message is dropped.
|
||||
case <-ctx.Done():
|
||||
log.Warnf("Tell to actor %s: message %s dropped "+
|
||||
"(caller context cancelled)", ref.actor.id,
|
||||
msg.MessageType())
|
||||
|
||||
// The actor itself has been stopped/terminated.
|
||||
case <-ref.actor.ctx.Done():
|
||||
// If the actor is terminated and has a DLO, send the message
|
||||
// there. Otherwise, it's dropped.
|
||||
ref.trySendToDLO(msg)
|
||||
// Use mailbox Send method which internally checks both contexts.
|
||||
if !ref.actor.mailbox.Send(ctx, env) {
|
||||
// Failed to send - check if actor terminated.
|
||||
if ref.actor.ctx.Err() != nil {
|
||||
ref.trySendToDLO(msg)
|
||||
}
|
||||
// Otherwise it was the caller's context that cancelled.
|
||||
}
|
||||
}
|
||||
|
||||
@@ -271,21 +223,16 @@ func (ref *actorRefImpl[M, R]) Ask(ctx context.Context, msg M) Future[R] {
|
||||
return promise.Future()
|
||||
}
|
||||
|
||||
select {
|
||||
// Attempt to send the message along with its promise to the actor's
|
||||
// mailbox.
|
||||
case ref.actor.mailbox <- envelope[M, R]{message: msg, promise: promise}:
|
||||
env := envelope[M, R]{message: msg, promise: promise}
|
||||
|
||||
// The context for the Ask operation was cancelled before the message
|
||||
// could be enqueued. Complete the promise with the context's error to
|
||||
// unblock the caller.
|
||||
case <-ctx.Done():
|
||||
promise.Complete(fn.Err[R](ctx.Err()))
|
||||
|
||||
// The actor's context was cancelled (e.g., actor stopped) while this
|
||||
// Ask operation was attempting to send (e.g., mailbox was full).
|
||||
case <-ref.actor.ctx.Done():
|
||||
promise.Complete(fn.Err[R](ErrActorTerminated))
|
||||
// Use mailbox Send method which internally checks both contexts.
|
||||
if !ref.actor.mailbox.Send(ctx, env) {
|
||||
// Determine the error based on what failed.
|
||||
if ref.actor.ctx.Err() != nil {
|
||||
promise.Complete(fn.Err[R](ErrActorTerminated))
|
||||
} else {
|
||||
promise.Complete(fn.Err[R](ctx.Err()))
|
||||
}
|
||||
}
|
||||
|
||||
// Return the future associated with the promise, allowing the caller to
|
||||
|
||||
Reference in new issue
Block a user