actor: add fundamental interfaces and concrete Actor impl
In this commit, we add the actual Actor implementation. We define a series of types and interfaces, that in concert, describe our actor. An actor has some ID, a reference (used to send messages to it), and also a set of defined messages that it'll accept. An actor can be implemented using a simple function if it's stateless. Otherwise, a struct can implement the Receive method, and handle its internal message passing and state that way.
This commit is contained in:
4 files changed
+888
No files matched your search
+324
@@ -0,0 +1,324 @@
|
||||
package actor
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
"github.com/lightningnetwork/lnd/fn/v2"
|
||||
)
|
||||
|
||||
// ActorConfig holds the configuration parameters for creating a new Actor.
|
||||
// It is generic over M (Message type) and R (Response type) to accommodate
|
||||
// the actor's specific behavior.
|
||||
type ActorConfig[M Message, R any] struct {
|
||||
// ID is the unique identifier for the actor.
|
||||
ID string
|
||||
|
||||
// Behavior defines how the actor responds to messages.
|
||||
Behavior ActorBehavior[M, R]
|
||||
|
||||
// DLO is a reference to the dead letter office for this actor system.
|
||||
// If nil, undeliverable messages during shutdown or due to a full
|
||||
// mailbox (if such logic were added) might be dropped.
|
||||
DLO ActorRef[Message, any]
|
||||
|
||||
// MailboxSize defines the buffer capacity of the actor's mailbox.
|
||||
MailboxSize int
|
||||
}
|
||||
|
||||
// envelope wraps a message with its associated promise. This allows the sender
|
||||
// of an "ask" message to await a response. If the promise is nil, it
|
||||
// signifies a "tell" operation (fire-and-forget).
|
||||
type envelope[M Message, R any] struct {
|
||||
message M
|
||||
promise Promise[R]
|
||||
}
|
||||
|
||||
// Actor represents a concrete actor implementation. It encapsulates a behavior,
|
||||
// manages its internal state implicitly through that behavior, and processes
|
||||
// messages from its mailbox sequentially in its own goroutine.
|
||||
type Actor[M Message, R any] struct {
|
||||
// id is the unique identifier for the actor.
|
||||
id string
|
||||
|
||||
// behavior defines how the actor responds to messages.
|
||||
behavior ActorBehavior[M, R]
|
||||
|
||||
// mailbox is the incoming message queue for the actor.
|
||||
mailbox chan envelope[M, R]
|
||||
|
||||
// ctx is the context governing the actor's lifecycle.
|
||||
ctx context.Context
|
||||
|
||||
// cancel is the function to cancel the actor's context.
|
||||
cancel context.CancelFunc
|
||||
|
||||
// dlo is a reference to the dead letter office for this actor system.
|
||||
dlo ActorRef[Message, any]
|
||||
|
||||
// startOnce ensures the actor's processing loop is started only once.
|
||||
startOnce sync.Once
|
||||
|
||||
// stopOnce ensures the actor's processing loop is stopped only once.
|
||||
stopOnce sync.Once
|
||||
|
||||
// ref is the cached ActorRef for this actor.
|
||||
ref ActorRef[M, R]
|
||||
}
|
||||
|
||||
// NewActor creates a new actor instance with the given ID and behavior.
|
||||
// It initializes the actor's internal structures but does not start its
|
||||
// message processing goroutine. The Start() method must be called to begin
|
||||
// processing messages.
|
||||
func NewActor[M Message, R any](cfg ActorConfig[M, R]) (*Actor[M, R],
|
||||
error) {
|
||||
|
||||
if cfg.ID == "" {
|
||||
return nil, ErrEmptyActorID
|
||||
}
|
||||
|
||||
if cfg.Behavior == nil {
|
||||
return nil, ErrNilBehavior
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
// Ensure MailboxSize has a sane default if not specified or zero. A
|
||||
// capacity of 0 would make the channel unbuffered, which is generally
|
||||
// not desired for actor mailboxes.
|
||||
mailboxCapacity := cfg.MailboxSize
|
||||
if mailboxCapacity <= 0 {
|
||||
// Default to a small capacity if an invalid one is given. This
|
||||
// could also come from a global constant.
|
||||
mailboxCapacity = 1
|
||||
}
|
||||
|
||||
actor := &Actor[M, R]{
|
||||
id: cfg.ID,
|
||||
behavior: cfg.Behavior,
|
||||
mailbox: make(chan envelope[M, R], mailboxCapacity),
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
dlo: cfg.DLO,
|
||||
}
|
||||
|
||||
// Create and cache the actor's own reference.
|
||||
actor.ref = &actorRefImpl[M, R]{
|
||||
actor: actor,
|
||||
}
|
||||
|
||||
return actor, nil
|
||||
}
|
||||
|
||||
// Start initiates the actor's message processing loop in a new goroutine. This
|
||||
// method should be called once after the actor is created.
|
||||
func (a *Actor[M, R]) Start() {
|
||||
a.startOnce.Do(func() {
|
||||
log.Infof("Actor %s: starting", a.id)
|
||||
|
||||
go a.process()
|
||||
})
|
||||
}
|
||||
|
||||
// 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)
|
||||
|
||||
// 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)
|
||||
|
||||
// Drain any remaining messages from the mailbox
|
||||
// without closing the channel.
|
||||
drained := 0
|
||||
for {
|
||||
select {
|
||||
case env := <-a.mailbox:
|
||||
drained++
|
||||
|
||||
// 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
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Stop signals the actor to terminate its processing loop and shut down.
|
||||
// This is achieved by cancelling the actor's internal context. The actor's
|
||||
// goroutine will exit once it detects the context cancellation.
|
||||
func (a *Actor[M, R]) Stop() {
|
||||
a.stopOnce.Do(func() {
|
||||
log.Infof("Actor %s: stopping", a.id)
|
||||
|
||||
a.cancel()
|
||||
})
|
||||
}
|
||||
|
||||
// actorRefImpl provides a concrete implementation of the ActorRef interface. It
|
||||
// holds a reference to the target Actor instance, enabling message sending.
|
||||
type actorRefImpl[M Message, R any] struct {
|
||||
actor *Actor[M, R]
|
||||
}
|
||||
|
||||
// Tell sends a message without waiting for a response. If the context is
|
||||
// cancelled before the message can be sent to the actor's mailbox, the message
|
||||
// may be dropped.
|
||||
//
|
||||
//nolint:ll
|
||||
func (ref *actorRefImpl[M, R]) Tell(ctx context.Context, msg M) {
|
||||
// If the actor's own context is already done, don't try to send.
|
||||
// Route to DLO if available.
|
||||
if ref.actor.ctx.Err() != nil {
|
||||
ref.trySendToDLO(msg)
|
||||
return
|
||||
}
|
||||
|
||||
select {
|
||||
// Message successfully enqueued in the actor's mailbox.
|
||||
case ref.actor.mailbox <- 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)
|
||||
}
|
||||
}
|
||||
|
||||
// Ask sends a message and returns a Future for the response. The Future will be
|
||||
// completed with the actor's reply or an error if the operation fails (e.g.,
|
||||
// context cancellation before send).
|
||||
//
|
||||
//nolint:ll
|
||||
func (ref *actorRefImpl[M, R]) Ask(ctx context.Context, msg M) Future[R] {
|
||||
// Create a new promise that will be fulfilled with the actor's response.
|
||||
promise := NewPromise[R]()
|
||||
|
||||
// If the actor's own context is already done, complete the promise with
|
||||
// ErrActorTerminated and return immediately. This is the primary guard
|
||||
// against trying to send to a stopped actor.
|
||||
if ref.actor.ctx.Err() != nil {
|
||||
promise.Complete(fn.Err[R](ErrActorTerminated))
|
||||
return promise.Future()
|
||||
}
|
||||
|
||||
// Check if the context is already done before attempting to send. This
|
||||
// ensures deterministic behavior and prevents a race where the message
|
||||
// could be enqueued even though the context was already cancelled.
|
||||
if ctx.Err() != nil {
|
||||
promise.Complete(fn.Err[R](ctx.Err()))
|
||||
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}:
|
||||
|
||||
// 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))
|
||||
}
|
||||
|
||||
// Return the future associated with the promise, allowing the caller to
|
||||
// await the response.
|
||||
return promise.Future()
|
||||
}
|
||||
|
||||
// trySendToDLO attempts to send the message to the actor's DLO if configured.
|
||||
func (ref *actorRefImpl[M, R]) trySendToDLO(msg M) {
|
||||
if ref.actor.dlo != nil {
|
||||
// Use context.Background() for sending to DLO as the
|
||||
// original context might be done or the operation
|
||||
// should not be bound by it.
|
||||
// This Tell to DLO is fire-and-forget.
|
||||
ref.actor.dlo.Tell(context.Background(), msg)
|
||||
}
|
||||
}
|
||||
|
||||
// ID returns the unique identifier for this actor.
|
||||
func (ref *actorRefImpl[M, R]) ID() string {
|
||||
return ref.actor.id
|
||||
}
|
||||
|
||||
// Ref returns an ActorRef for this actor. This allows clients to interact with
|
||||
// the actor (send messages) without having direct access to the Actor struct
|
||||
// itself, promoting encapsulation and location transparency.
|
||||
func (a *Actor[M, R]) Ref() ActorRef[M, R] {
|
||||
return a.ref
|
||||
}
|
||||
|
||||
// TellRef returns a TellOnlyRef for this actor. This allows clients to send
|
||||
// messages to the actor using only the "tell" pattern (fire-and-forget),
|
||||
// without having access to "ask" capabilities.
|
||||
func (a *Actor[M, R]) TellRef() TellOnlyRef[M] {
|
||||
return a.ref
|
||||
}
|
||||
@@ -0,0 +1,446 @@
|
||||
package actor
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/lightningnetwork/lnd/fn/v2"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// testMsg is a simple message type for testing. It embeds BaseMessage to
|
||||
// satisfy the actor.Message interface.
|
||||
type testMsg struct {
|
||||
BaseMessage
|
||||
data string
|
||||
|
||||
replyChan chan string
|
||||
}
|
||||
|
||||
// MessageType returns the type name of the message.
|
||||
func (m *testMsg) MessageType() string {
|
||||
return "testMsg"
|
||||
}
|
||||
|
||||
// newTestMsg creates a new test message.
|
||||
func newTestMsg(data string) *testMsg {
|
||||
return &testMsg{data: data}
|
||||
}
|
||||
|
||||
// newTestMsgWithReply creates a new test message that includes a reply channel.
|
||||
// This can be used by test behaviors to send data back to the test
|
||||
// synchronously, especially for Tell operations.
|
||||
func newTestMsgWithReply(data string, replyChan chan string) *testMsg {
|
||||
return &testMsg{data: data, replyChan: replyChan}
|
||||
}
|
||||
|
||||
// echoBehavior is a simple actor behavior that processes *testMsg messages. It
|
||||
// stores the last message's data and, for Ask, echoes it back. For Tell, if
|
||||
// replyChan is set in testMsg, it sends data back on it.
|
||||
type echoBehavior struct {
|
||||
lastMsgData atomic.Value
|
||||
processingDelay time.Duration
|
||||
t *testing.T
|
||||
}
|
||||
|
||||
// newEchoBehavior creates a new echoBehavior.
|
||||
func newEchoBehavior(t *testing.T, delay time.Duration) *echoBehavior {
|
||||
return &echoBehavior{t: t, processingDelay: delay}
|
||||
}
|
||||
|
||||
// Receive handles incoming messages. It simulates work if processingDelay is
|
||||
// set, stores the message data, and responds for Ask operations or via
|
||||
// replyChan for Tell.
|
||||
func (b *echoBehavior) Receive(_ context.Context,
|
||||
msg *testMsg) fn.Result[string] {
|
||||
|
||||
if b.processingDelay > 0 {
|
||||
time.Sleep(b.processingDelay)
|
||||
}
|
||||
|
||||
b.lastMsgData.Store(msg.data)
|
||||
|
||||
if msg.replyChan != nil {
|
||||
// Attempt to send the data on the reply channel, but quit if
|
||||
// it takes longer than 1 second (e.g., channel unbuffered
|
||||
// and no receiver).
|
||||
select {
|
||||
case msg.replyChan <- msg.data:
|
||||
case <-time.After(time.Second):
|
||||
b.t.Logf("warning: replyChan send timed out")
|
||||
}
|
||||
}
|
||||
|
||||
return fn.Ok(fmt.Sprintf("echo: %s", msg.data))
|
||||
}
|
||||
|
||||
// GetLastMsgData retrieves the data from the last message processed.
|
||||
func (b *echoBehavior) GetLastMsgData() (string, bool) {
|
||||
val := b.lastMsgData.Load()
|
||||
if val == nil {
|
||||
return "", false
|
||||
}
|
||||
data, ok := val.(string)
|
||||
return data, ok
|
||||
}
|
||||
|
||||
// errorBehavior is an actor behavior that always returns a predefined error
|
||||
// upon receiving a message.
|
||||
type errorBehavior struct {
|
||||
err error
|
||||
}
|
||||
|
||||
// newErrorBehavior creates a new errorBehavior.
|
||||
func newErrorBehavior(err error) *errorBehavior {
|
||||
return &errorBehavior{err: err}
|
||||
}
|
||||
|
||||
// Receive always returns the configured error.
|
||||
func (b *errorBehavior) Receive(_ context.Context,
|
||||
_ *testMsg) fn.Result[string] {
|
||||
|
||||
return fn.Err[string](b.err)
|
||||
}
|
||||
|
||||
// blockingBehavior is an actor behavior that blocks until its actorCtx is done.
|
||||
type blockingBehavior struct{}
|
||||
|
||||
// Receive blocks until the actor's context is cancelled, then returns the
|
||||
// context's error.
|
||||
func (b *blockingBehavior) Receive(actorCtx context.Context,
|
||||
_ *testMsg) fn.Result[string] {
|
||||
|
||||
<-actorCtx.Done()
|
||||
return fn.Err[string](actorCtx.Err())
|
||||
}
|
||||
|
||||
// deadLetterTestMsg is a distinct message type used for testing DLO
|
||||
// interactions.
|
||||
type deadLetterTestMsg struct {
|
||||
BaseMessage
|
||||
id string
|
||||
}
|
||||
|
||||
// MessageType returns the type name of the message.
|
||||
func (m *deadLetterTestMsg) MessageType() string {
|
||||
return "deadLetterTestMsg"
|
||||
}
|
||||
|
||||
// deadLetterObserverBehavior is a behavior for a test Dead Letter Office actor.
|
||||
// It records all messages sent to it, allowing tests to verify DLO
|
||||
// interactions.
|
||||
type deadLetterObserverBehavior struct {
|
||||
mu sync.Mutex
|
||||
receivedMsgs []Message
|
||||
}
|
||||
|
||||
// newDeadLetterObserverBehavior creates a new deadLetterObserverBehavior.
|
||||
func newDeadLetterObserverBehavior() *deadLetterObserverBehavior {
|
||||
return &deadLetterObserverBehavior{
|
||||
receivedMsgs: make([]Message, 0),
|
||||
}
|
||||
}
|
||||
|
||||
// Receive records the incoming message and returns a successful result.
|
||||
func (b *deadLetterObserverBehavior) Receive(_ context.Context,
|
||||
msg Message) fn.Result[any] {
|
||||
|
||||
b.mu.Lock()
|
||||
b.receivedMsgs = append(b.receivedMsgs, msg)
|
||||
b.mu.Unlock()
|
||||
|
||||
return fn.Ok[any](nil)
|
||||
}
|
||||
|
||||
// GetReceivedMsgs returns a copy of all messages received by this DLO.
|
||||
func (b *deadLetterObserverBehavior) GetReceivedMsgs() []Message {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
|
||||
msgs := make([]Message, len(b.receivedMsgs))
|
||||
copy(msgs, b.receivedMsgs)
|
||||
|
||||
return msgs
|
||||
}
|
||||
|
||||
// actorTestHarness provides helper methods for setting up actors in tests. It
|
||||
// manages a dedicated DLO for actors created through it.
|
||||
type actorTestHarness struct {
|
||||
t *testing.T
|
||||
dlo *Actor[Message, any]
|
||||
dloBeh *deadLetterObserverBehavior
|
||||
}
|
||||
|
||||
// newActorTestHarness sets up a test harness with a dedicated DLO. The DLO is
|
||||
// automatically stopped when the test cleans up.
|
||||
func newActorTestHarness(t *testing.T) *actorTestHarness {
|
||||
t.Helper()
|
||||
|
||||
dloBeh := newDeadLetterObserverBehavior()
|
||||
dloCfg := ActorConfig[Message, any]{
|
||||
ID: "test-dlo-" + t.Name(),
|
||||
Behavior: dloBeh,
|
||||
DLO: nil,
|
||||
MailboxSize: 10,
|
||||
}
|
||||
dloActor, err := NewActor[Message, any](dloCfg)
|
||||
require.NoError(t, err)
|
||||
dloActor.Start()
|
||||
|
||||
t.Cleanup(dloActor.Stop)
|
||||
|
||||
return &actorTestHarness{
|
||||
t: t,
|
||||
dlo: dloActor,
|
||||
dloBeh: dloBeh,
|
||||
}
|
||||
}
|
||||
|
||||
// newActor creates, starts, and registers a new actor for cleanup. The actor
|
||||
// will use the harness's DLO.
|
||||
func (h *actorTestHarness) newActor(id string,
|
||||
beh ActorBehavior[*testMsg, string],
|
||||
mailboxSize int) *Actor[*testMsg, string] {
|
||||
|
||||
h.t.Helper()
|
||||
|
||||
cfg := ActorConfig[*testMsg, string]{
|
||||
ID: id,
|
||||
Behavior: beh,
|
||||
DLO: h.dlo.Ref(),
|
||||
MailboxSize: mailboxSize,
|
||||
}
|
||||
actor, err := NewActor(cfg)
|
||||
require.NoError(h.t, err)
|
||||
actor.Start()
|
||||
|
||||
h.t.Cleanup(actor.Stop)
|
||||
|
||||
return actor
|
||||
}
|
||||
|
||||
// assertDLOMessage checks that the DLO eventually receives a specific message.
|
||||
func (h *actorTestHarness) assertDLOMessage(expectedMsg Message) {
|
||||
h.t.Helper()
|
||||
require.Eventually(h.t, func() bool {
|
||||
msgs := h.dloBeh.GetReceivedMsgs()
|
||||
for _, m := range msgs {
|
||||
if reflect.DeepEqual(m, expectedMsg) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}, time.Second, 10*time.Millisecond,
|
||||
"dLO did not receive expected message: %v", expectedMsg,
|
||||
)
|
||||
}
|
||||
|
||||
// assertNoDLOMessages checks that the DLO has not received any messages.
|
||||
func (h *actorTestHarness) assertNoDLOMessages() {
|
||||
h.t.Helper()
|
||||
|
||||
// Allow a very brief moment for any async DLO sends to occur.
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
|
||||
msgs := h.dloBeh.GetReceivedMsgs()
|
||||
|
||||
require.Empty(h.t, msgs, "dLO received unexpected messages")
|
||||
}
|
||||
|
||||
// TestActorNewActorIDAndRefs verifies that NewActor correctly initializes an
|
||||
// actor's ID and provides functional ActorRef and TellOnlyRef instances.
|
||||
func TestActorNewActorIDAndRefs(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
h := newActorTestHarness(t)
|
||||
actorID := "test-actor-1"
|
||||
beh := newEchoBehavior(t, 0)
|
||||
actor := h.newActor(actorID, beh, 1)
|
||||
|
||||
require.Equal(t, actorID, actor.Ref().ID(), "actorRef ID mismatch")
|
||||
require.Equal(
|
||||
t, actorID, actor.TellRef().ID(), "tellOnlyRef ID mismatch",
|
||||
)
|
||||
require.NotNil(t, actor.Ref(), "actorRef should not be nil")
|
||||
require.NotNil(t, actor.TellRef(), "tellOnlyRef should not be nil")
|
||||
}
|
||||
|
||||
// TestActorStartStop verifies the basic lifecycle of an actor: starting,
|
||||
// processing messages, and stopping.
|
||||
func TestActorStartStop(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
h := newActorTestHarness(t)
|
||||
beh := newEchoBehavior(t, 0)
|
||||
actor := h.newActor("test-actor-lifecycle", beh, 1)
|
||||
|
||||
// Actor should be running and process a message.
|
||||
msgData := "hello"
|
||||
replyChan := make(chan string, 1)
|
||||
actor.Ref().Tell(
|
||||
context.Background(), newTestMsgWithReply(msgData, replyChan),
|
||||
)
|
||||
|
||||
received, err := fn.RecvOrTimeout(replyChan, 100*time.Millisecond)
|
||||
require.NoError(t, err, "timed out waiting for actor to process message")
|
||||
require.Equal(
|
||||
t, msgData, received, "actor did not process message before stop",
|
||||
)
|
||||
|
||||
actor.Stop()
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
// Try sending another message; it should ideally not be processed or go
|
||||
// to DLO.
|
||||
msgDataAfterStop := "message-after-stop"
|
||||
replyChanAfterStop := make(chan string, 1)
|
||||
actor.Ref().Tell(
|
||||
context.Background(),
|
||||
newTestMsgWithReply(msgDataAfterStop, replyChanAfterStop),
|
||||
)
|
||||
|
||||
// We expect a timeout here, meaning the message was not processed by
|
||||
// the echoBehavior's replyChan.
|
||||
_, err = fn.RecvOrTimeout(replyChanAfterStop, 100*time.Millisecond)
|
||||
// err == nil would mean a message was received, meaning the actor
|
||||
// processed it after Stop().
|
||||
require.Error(t, err, "actor processed message after Stop()")
|
||||
require.ErrorContains(t, err, "timeout hit")
|
||||
|
||||
h.assertDLOMessage(
|
||||
&testMsg{data: msgDataAfterStop, replyChan: replyChanAfterStop},
|
||||
)
|
||||
}
|
||||
|
||||
// TestActorTellBasic verifies that a message sent via Tell is processed by the
|
||||
// actor's behavior.
|
||||
func TestActorTellBasic(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
h := newActorTestHarness(t)
|
||||
beh := newEchoBehavior(t, 0)
|
||||
actor := h.newActor("test-actor-tell", beh, 1)
|
||||
|
||||
msgData := "tell-message"
|
||||
replyChan := make(chan string, 1)
|
||||
actor.Ref().Tell(
|
||||
context.Background(), newTestMsgWithReply(msgData, replyChan),
|
||||
)
|
||||
|
||||
receivedTell, errTell := fn.RecvOrTimeout(replyChan, 100*time.Millisecond)
|
||||
require.NoError(t, errTell, "timed out waiting for Tell message processing")
|
||||
require.Equal(
|
||||
t, msgData, receivedTell, "behavior did not receive Tell message data",
|
||||
)
|
||||
|
||||
lastData, ok := beh.GetLastMsgData()
|
||||
require.True(t, ok, "last message data not set in behavior")
|
||||
require.Equal(t, msgData, lastData, "last message data mismatch")
|
||||
h.assertNoDLOMessages()
|
||||
}
|
||||
|
||||
// TestActorAskSuccess verifies that a message sent via Ask is processed, and
|
||||
// the returned Future is completed with the behavior's successful result.
|
||||
func TestActorAskSuccess(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
h := newActorTestHarness(t)
|
||||
beh := newEchoBehavior(t, 0)
|
||||
actor := h.newActor("test-actor-ask-success", beh, 1)
|
||||
|
||||
msgData := "ask-message"
|
||||
future := actor.Ref().Ask(context.Background(), newTestMsg(msgData))
|
||||
|
||||
result := future.Await(context.Background())
|
||||
require.False(t, result.IsErr(), "ask returned an error: %v", result.Err())
|
||||
|
||||
result.WhenOk(func(val string) {
|
||||
expectedReply := fmt.Sprintf("echo: %s", msgData)
|
||||
require.Equal(t, expectedReply, val, "ask response mismatch")
|
||||
})
|
||||
|
||||
lastData, ok := beh.GetLastMsgData()
|
||||
require.True(t, ok, "last message data not set in behavior")
|
||||
require.Equal(t, msgData, lastData, "last message data mismatch")
|
||||
h.assertNoDLOMessages()
|
||||
}
|
||||
|
||||
// TestActorAskErrorBehavior verifies that if an actor's behavior returns an
|
||||
// error, the Future from an Ask call is completed with that error.
|
||||
func TestActorAskErrorBehavior(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
h := newActorTestHarness(t)
|
||||
expectedErr := errors.New("behavior error")
|
||||
beh := newErrorBehavior(expectedErr)
|
||||
actor := h.newActor("test-actor-ask-error", beh, 1)
|
||||
|
||||
future := actor.Ref().Ask(
|
||||
context.Background(), newTestMsg("ask-error-test"),
|
||||
)
|
||||
|
||||
result := future.Await(context.Background())
|
||||
require.True(t, result.IsErr(), "ask should have returned an error")
|
||||
require.ErrorIs(t, result.Err(), expectedErr, "ask error mismatch")
|
||||
|
||||
h.assertNoDLOMessages()
|
||||
}
|
||||
|
||||
// TestFunctionBehaviorFromSimple verifies that FunctionBehaviorFromSimple
|
||||
// correctly adapts a simple (msg) -> (result, error) function into an
|
||||
// ActorBehavior, handling both success and error cases.
|
||||
func TestFunctionBehaviorFromSimple(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
t.Run("success", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
h := newActorTestHarness(t)
|
||||
|
||||
beh := FunctionBehaviorFromSimple(
|
||||
func(msg *testMsg) (string, error) {
|
||||
return "simple: " + msg.data, nil
|
||||
},
|
||||
)
|
||||
actor := h.newActor("test-simple-success", beh, 1)
|
||||
|
||||
future := actor.Ref().Ask(
|
||||
context.Background(), newTestMsg("hello"),
|
||||
)
|
||||
result := future.Await(context.Background())
|
||||
require.False(
|
||||
t, result.IsErr(),
|
||||
"expected success, got: %v", result.Err(),
|
||||
)
|
||||
result.WhenOk(func(val string) {
|
||||
require.Equal(t, "simple: hello", val)
|
||||
})
|
||||
})
|
||||
|
||||
t.Run("error", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
h := newActorTestHarness(t)
|
||||
|
||||
expectedErr := errors.New("simple behavior error")
|
||||
beh := FunctionBehaviorFromSimple(
|
||||
func(msg *testMsg) (string, error) {
|
||||
return "", expectedErr
|
||||
},
|
||||
)
|
||||
actor := h.newActor("test-simple-error", beh, 1)
|
||||
|
||||
future := actor.Ref().Ask(
|
||||
context.Background(), newTestMsg("hello"),
|
||||
)
|
||||
result := future.Await(context.Background())
|
||||
require.True(t, result.IsErr())
|
||||
require.ErrorIs(t, result.Err(), expectedErr)
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
package actor
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/lightningnetwork/lnd/fn/v2"
|
||||
)
|
||||
|
||||
// ActorFunc is a function type that represents an actor which functions purely
|
||||
// based on a simple function processor.
|
||||
type ActorFunc[M Message, R any] func(context.Context, M) fn.Result[R]
|
||||
|
||||
// FunctionBehavior adapts a function to the ActorBehavior interface.
|
||||
type FunctionBehavior[M Message, R any] struct {
|
||||
fn ActorFunc[M, R]
|
||||
}
|
||||
|
||||
// NewFunctionBehavior creates a behavior from a function.
|
||||
func NewFunctionBehavior[M Message, R any](
|
||||
fn ActorFunc[M, R]) *FunctionBehavior[M, R] {
|
||||
|
||||
return &FunctionBehavior[M, R]{fn: fn}
|
||||
}
|
||||
|
||||
// Receive implements ActorBehavior interface for the function.
|
||||
//
|
||||
// TODO(roasbeef): just base it off the function direct instead?
|
||||
func (b *FunctionBehavior[M, R]) Receive(ctx context.Context,
|
||||
msg M) fn.Result[R] {
|
||||
|
||||
return b.fn(ctx, msg)
|
||||
}
|
||||
|
||||
// FunctionBehaviorFromSimple adapts a simpler function to the ActorBehavior
|
||||
// interface.
|
||||
func FunctionBehaviorFromSimple[M Message, R any](
|
||||
sFunc func(M) (R, error)) *FunctionBehavior[M, R] {
|
||||
|
||||
return NewFunctionBehavior(
|
||||
func(ctx context.Context, msg M) fn.Result[R] {
|
||||
val, err := sFunc(msg)
|
||||
return fn.NewResult(val, err)
|
||||
},
|
||||
)
|
||||
}
|
||||
@@ -7,6 +7,44 @@ import (
|
||||
"github.com/lightningnetwork/lnd/fn/v2"
|
||||
)
|
||||
|
||||
// ErrActorTerminated indicates that an operation failed because the target
|
||||
// actor was terminated or in the process of shutting down.
|
||||
var ErrActorTerminated = fmt.Errorf("actor terminated")
|
||||
|
||||
// ErrEmptyActorID is returned when an actor is created with an empty ID.
|
||||
var ErrEmptyActorID = fmt.Errorf("actor ID must not be empty")
|
||||
|
||||
// ErrNilBehavior is returned when an actor is created with a nil behavior.
|
||||
var ErrNilBehavior = fmt.Errorf("actor behavior must not be nil")
|
||||
|
||||
// ErrDuplicateActorID is returned when attempting to register an actor with an
|
||||
// ID that is already in use within the actor system.
|
||||
var ErrDuplicateActorID = fmt.Errorf("actor ID already registered")
|
||||
|
||||
// BaseMessage is a helper struct that can be embedded in message types defined
|
||||
// outside the actor package to satisfy the Message interface's unexported
|
||||
// messageMarker method.
|
||||
type BaseMessage struct{}
|
||||
|
||||
// messageMarker implements the unexported method for the Message interface,
|
||||
// allowing types that embed BaseMessage to satisfy the Message interface.
|
||||
func (BaseMessage) messageMarker() {}
|
||||
|
||||
// Message is a sealed interface for actor messages. Actors will receive
|
||||
// messages conforming to this interface. The interface is "sealed" by the
|
||||
// unexported messageMarker method, meaning only types that can satisfy it
|
||||
// (e.g., by embedding BaseMessage or being in the same package) can be
|
||||
// Messages.
|
||||
type Message interface {
|
||||
// messageMarker is a private method that makes this a sealed interface
|
||||
// (see BaseMessage for embedding).
|
||||
messageMarker()
|
||||
|
||||
// MessageType returns the type name of the message for
|
||||
// routing/filtering.
|
||||
MessageType() string
|
||||
}
|
||||
|
||||
// Future represents the result of an asynchronous computation. It allows
|
||||
// consumers to wait for the result (Await), apply transformations upon
|
||||
// completion (ThenApply), or register a callback to be executed when the
|
||||
@@ -44,3 +82,38 @@ type Promise[T any] interface {
|
||||
// complete it), and false if the future had already been completed.
|
||||
Complete(result fn.Result[T]) bool
|
||||
}
|
||||
|
||||
// TellOnlyRef is a reference to an actor that only supports "tell" operations.
|
||||
// This is useful for scenarios where only fire-and-forget message passing is
|
||||
// needed, or to restrict capabilities.
|
||||
type TellOnlyRef[M Message] interface {
|
||||
// Tell sends a message without waiting for a response. If the
|
||||
// context is cancelled before the message can be sent to the actor's
|
||||
// mailbox, the message may be dropped.
|
||||
Tell(ctx context.Context, msg M)
|
||||
|
||||
// ID returns the unique identifier for this actor.
|
||||
ID() string
|
||||
}
|
||||
|
||||
// ActorRef is a reference to an actor that supports both "tell" and "ask"
|
||||
// operations. It embeds TellOnlyRef and adds the Ask method for
|
||||
// request-response interactions.
|
||||
type ActorRef[M Message, R any] interface {
|
||||
TellOnlyRef[M]
|
||||
|
||||
// Ask sends a message and returns a Future for the response.
|
||||
// The Future will be completed with the actor's reply or an error
|
||||
// if the operation fails (e.g., context cancellation before send).
|
||||
Ask(ctx context.Context, msg M) Future[R]
|
||||
}
|
||||
|
||||
// ActorBehavior defines the logic for how an actor processes incoming messages.
|
||||
// It is a strategy interface that encapsulates the actor's reaction to
|
||||
// messages.
|
||||
type ActorBehavior[M Message, R any] interface {
|
||||
// Receive processes a message and returns a Result. The provided
|
||||
// context is the actor's internal context, which can be used to
|
||||
// detect actor shutdown requests.
|
||||
Receive(actorCtx context.Context, msg M) fn.Result[R]
|
||||
}
|
||||
Reference in new issue
Block a user