Repository: Tochemey/goakt
Pull request #96 (2023-08-07 03:19:36)

feat: add api tests


Patch:
diff --git a/Earthfile b/Earthfile
index 12f8c96..d8aac86 100644
--- a/Earthfile
+++ b/Earthfile
@@ -1,4 +1,6 @@
 VERSION 0.7
+PROJECT tochemey26/goakt
+
 
 FROM tochemey/docker-go:1.20.6-0.9.1
 
diff --git a/actors/pid.go b/actors/pid.go
index 6501a0e..f162e58 100644
--- a/actors/pid.go
+++ b/actors/pid.go
@@ -129,19 +129,20 @@ type pid struct {
 	// specifies at what point in time to passivate the actor.
 	// when the actor is passivated it is stopped which means it does not consume
 	// any further resources like memory and cpu. The default value is 120 seconds
-	passivateAfter *atomic.Duration
+	passivateAfter time.Duration
 
 	// specifies how long the sender of a mail should wait to receive a reply
 	// when using SendReply. The default value is 5s
-	sendReplyTimeout *atomic.Duration
+	sendReplyTimeout time.Duration
 
 	// specifies the maximum of retries to attempt when the actor
 	// initialization fails. The default value is 5 seconds
-	initMaxRetries *atomic.Int32
+	initMaxRetries int
 
-	// shutdownTimeout specifies the graceful shutdown timeout
+	// shutdownTimeout specifies how long to wait for the remaining messages
+	// in the actor mailbox to be processed before completing the shutdown process
 	// the default value is 5 seconds
-	shutdownTimeout *atomic.Duration
+	shutdownTimeout time.Duration
 
 	// specifies the actor mailbox
 	mailbox chan ReceiveContext
@@ -197,10 +198,10 @@ func newPID(ctx context.Context, actorPath *Path, actor Actor, opts ...pidOption
 		Actor:                  actor,
 		isRunning:              atomic.NewBool(false),
 		lastProcessingTime:     atomic.Time{},
-		passivateAfter:         atomic.NewDuration(DefaultPassivationTimeout),
-		sendReplyTimeout:       atomic.NewDuration(DefaultReplyTimeout),
-		shutdownTimeout:        atomic.NewDuration(DefaultShutdownTimeout),
-		initMaxRetries:         atomic.NewInt32(DefaultInitMaxRetries),
+		passivateAfter:         DefaultPassivationTimeout,
+		sendReplyTimeout:       DefaultReplyTimeout,
+		shutdownTimeout:        DefaultShutdownTimeout,
+		initMaxRetries:         DefaultInitMaxRetries,
 		mailbox:                make(chan ReceiveContext, defaultMailboxSize),
 		shutdownSignal:         make(chan Unit, 1),
 		haltPassivationLnr:     make(chan Unit, 1),
@@ -228,7 +229,7 @@ func newPID(ctx context.Context, actorPath *Path, actor Actor, opts ...pidOption
 	// init processing public
 	go pid.receive()
 	// init the passivation listener loop iff passivation is set
-	if pid.passivateAfter.Load() > 0 {
+	if pid.passivateAfter > 0 {
 		go pid.passivationListener()
 	}
 
@@ -329,7 +330,7 @@ func (p *pid) Restart(ctx context.Context) error {
 	// init processing public
 	go p.receive()
 	// init the passivation listener loop iff passivation is set
-	if p.passivateAfter.Load() > 0 {
+	if p.passivateAfter > 0 {
 		go p.passivationListener()
 	}
 	// increment the restart counter
@@ -372,13 +373,13 @@ func (p *pid) SpawnChild(ctx context.Context, name string, actor Actor) (PID, er
 	cid := newPID(ctx,
 		childActorPath,
 		actor,
-		withInitMaxRetries(int(p.initMaxRetries.Load())),
-		withPassivationAfter(p.passivateAfter.Load()),
-		withSendReplyTimeout(p.sendReplyTimeout.Load()),
+		withInitMaxRetries(p.initMaxRetries),
+		withPassivationAfter(p.passivateAfter),
+		withSendReplyTimeout(p.sendReplyTimeout),
 		withCustomLogger(p.logger),
 		withActorSystem(p.system),
 		withSupervisorStrategy(p.supervisorStrategy),
-		withShutdownTimeout(p.shutdownTimeout.Load()))
+		withShutdownTimeout(p.shutdownTimeout))
 
 	// add the pid to the map
 	p.children.Set(cid)
@@ -463,7 +464,7 @@ func (p *pid) Ask(ctx context.Context, to PID, message proto.Message) (response
 	to.doReceive(context)
 
 	// await patiently to receive the response from the actor
-	for await := time.After(p.sendReplyTimeout.Load()); ; {
+	for await := time.After(p.sendReplyTimeout); ; {
 		select {
 		case response = <-context.response:
 			return
@@ -651,19 +652,13 @@ func (p *pid) Shutdown(ctx context.Context) error {
 	}
 
 	// stop the passivation listener
-	if p.passivateAfter.Load() > 0 {
-		// add some debug logging
-		p.logger.Debug("sending a signal to stop passivation listener....")
-		p.haltPassivationLnr <- Unit{}
-	}
+	p.haltPassivationLnr <- Unit{}
 
 	// signal we are shutting down to stop processing messages
 	p.shutdownSignal <- Unit{}
 
-	// stop receiving and sending messages
-	p.isRunning.Store(false)
-	// stop all the child actors
-	p.freeChildren(ctx)
+	// stop the actor PID
+	p.stop(ctx)
 
 	// add some logging
 	p.logger.Infof("Shutdown process is on going for actor=%s...", p.ActorPath().String())
@@ -749,7 +744,7 @@ func (p *pid) init(ctx context.Context) {
 	// add some logging info
 	p.logger.Info("Initialization process has started...")
 	// create the exponential backoff object
-	expoBackoff := backoff.WithMaxRetries(backoff.NewExponentialBackOff(), uint64(p.initMaxRetries.Load()))
+	expoBackoff := backoff.WithMaxRetries(backoff.NewExponentialBackOff(), uint64(p.initMaxRetries))
 	// init the actor initialization receive
 	err := backoff.Retry(func() error {
 		return p.Actor.PreStart(ctx)
@@ -776,15 +771,18 @@ func (p *pid) reset() {
 	defer p.mu.Unlock()
 	// reset the various settings of the PID
 	p.lastProcessingTime = atomic.Time{}
-	p.passivateAfter.Store(DefaultPassivationTimeout)
-	p.sendReplyTimeout.Store(DefaultReplyTimeout)
-	p.shutdownTimeout.Store(DefaultShutdownTimeout)
-	p.initMaxRetries.Store(DefaultInitMaxRetries)
-	p.panicCounter.Store(0)
-	p.receivedMessageCounter.Store(0)
-	p.lastProcessingDuration.Store(0)
-	p.restartCounter.Store(0)
-	p.mailboxSizeCounter.Store(0)
+	p.passivateAfter = DefaultPassivationTimeout
+	p.sendReplyTimeout = DefaultReplyTimeout
+	p.shutdownTimeout = DefaultShutdownTimeout
+	p.initMaxRetries = DefaultInitMaxRetries
+	p.mailbox = make(chan ReceiveContext, defaultMailboxSize)
+	p.shutdownSignal = make(chan Unit, 1)
+	p.haltPassivationLnr = make(chan Unit, 1)
+	p.panicCounter = atomic.NewUint64(0)
+	p.receivedMessageCounter = atomic.NewUint64(0)
+	p.lastProcessingDuration = atomic.NewDuration(0)
+	p.restartCounter = atomic.NewUint64(0)
+	p.mailboxSizeCounter = atomic.NewUint64(0)
 	p.children = newPIDMap(10)
 	p.watchMen = slices.NewConcurrentSlice[*WatchMan]()
 	p.telemetry = telemetry.New()
@@ -801,9 +799,9 @@ func (p *pid) reset() {
 
 func (p *pid) resetBehaviorStack() {
 	// reset the behavior
-	p.behaviorStack.Clear()
-	// set it to the receiver method
-	p.behaviorStack.Push(p.Receive)
+	behaviorStack := NewBehaviorStack()
+	behaviorStack.Push(p.Receive)
+	p.behaviorStack = behaviorStack
 }
 
 func (p *pid) freeChildren(ctx context.Context) {
@@ -825,6 +823,16 @@ func (p *pid) freeChildren(ctx context.Context) {
 	}
 }
 
+// stop stops the PID
+func (p *pid) stop(ctx context.Context) {
+	// add a span context
+	ctx, span := telemetry.SpanContext(ctx, "Stop")
+	defer span.End()
+	p.isRunning.Store(false)
+	// stop all the child actors
+	p.freeChildren(ctx)
+}
+
 // Behaviors returns the behavior stack
 func (p *pid) behaviors() BehaviorStack {
 	p.mu.Lock()
@@ -924,9 +932,6 @@ func (p *pid) handleReceived(received ReceiveContext) {
 			p.panicCounter.Inc()
 		}
 	}()
-	// synchronize the receiving call
-	p.mu.Lock()
-	defer p.mu.Unlock()
 	// send the message to the current actor behavior
 	if behavior, ok := p.behaviorStack.Peek(); ok {
 		behavior(received)
@@ -980,7 +985,7 @@ func (p *pid) supervise(cid PID, watcher *WatchMan) {
 func (p *pid) passivationListener() {
 	p.logger.Info("start the passivation listener...")
 	// create the ticker
-	ticker := time.NewTicker(p.passivateAfter.Load())
+	ticker := time.NewTicker(p.passivateAfter)
 	// create the stop ticker signal
 	tickerStopSig := make(chan Unit, 1)
 
@@ -992,7 +997,7 @@ func (p *pid) passivationListener() {
 				// check whether the actor is idle or not
 				idleTime := time.Since(p.lastProcessingTime.Load())
 				// check whether the actor is idle
-				if idleTime >= p.passivateAfter.Load() {
+				if idleTime >= p.passivateAfter {
 					// set the done channel to stop the ticker
 					tickerStopSig <- Unit{}
 					return
@@ -1006,10 +1011,8 @@ func (p *pid) passivationListener() {
 	}()
 	// wait for the stop signal to stop the ticker
 	<-tickerStopSig
-
 	// stop the ticker
 	ticker.Stop()
-
 	// only passivate when actor is alive
 	if !p.IsRunning() {
 		// add some logging info
@@ -1019,25 +1022,23 @@ func (p *pid) passivationListener() {
 
 	// add some logging info
 	p.logger.Infof("Passivation mode has been triggered for actor=%s...", p.ActorPath().String())
+	// passivate the actor
+	p.passivate()
+}
 
-	// signal we are shutting down to stop processing messages
-	p.shutdownSignal <- Unit{}
-
-	// stop receiving and sending messages
-	p.isRunning.Store(false)
-
+func (p *pid) passivate() {
 	// create a context
 	ctx := context.Background()
-
-	// stop all the child actors
-	p.freeChildren(ctx)
+	// stop processing messages
+	p.shutdownSignal <- Unit{}
+	// stop the passivation
+	p.haltPassivationLnr <- Unit{}
+	// stop the actor PID
+	p.stop(ctx)
 	// perform some cleanup with the actor
 	if err := p.Actor.PostStop(ctx); err != nil {
 		panic(err)
 	}
-
-	// reset
-	p.reset()
 }
 
 // interceptor create an interceptor based upon the telemetry provided
diff --git a/actors/pid_option.go b/actors/pid_option.go
index dca404c..827e4e5 100644
--- a/actors/pid_option.go
+++ b/actors/pid_option.go
@@ -5,7 +5,6 @@ import (
 
 	"github.com/tochemey/goakt/log"
 	"github.com/tochemey/goakt/pkg/telemetry"
-	"go.uber.org/atomic"
 )
 
 // pidOption represents the pid
@@ -14,7 +13,7 @@ type pidOption func(pid *pid)
 // withPassivationAfter sets the actor passivation time
 func withPassivationAfter(duration time.Duration) pidOption {
 	return func(pid *pid) {
-		pid.passivateAfter = atomic.NewDuration(duration)
+		pid.passivateAfter = duration
 	}
 }
 
@@ -22,14 +21,14 @@ func withPassivationAfter(duration time.Duration) pidOption {
 // in a receive-reply pattern
 func withSendReplyTimeout(timeout time.Duration) pidOption {
 	return func(pid *pid) {
-		pid.sendReplyTimeout = atomic.NewDuration(timeout)
+		pid.sendReplyTimeout = timeout
 	}
 }
 
 // withInitMaxRetries sets the number of times to retry an actor init process
 func withInitMaxRetries(max int) pidOption {
 	return func(pid *pid) {
-		pid.initMaxRetries = atomic.NewInt32(int32(max))
+		pid.initMaxRetries = max
 	}
 }
 
@@ -58,14 +57,14 @@ func withSupervisorStrategy(strategy StrategyDirective) pidOption {
 // withShutdownTimeout sets the shutdown timeout
 func withShutdownTimeout(duration time.Duration) pidOption {
 	return func(pid *pid) {
-		pid.shutdownTimeout = atomic.NewDuration(duration)
+		pid.shutdownTimeout = duration
 	}
 }
 
 // withNoPassivation disable passivation
 func withPassivationDisabled() pidOption {
 	return func(pid *pid) {
-		pid.passivateAfter = atomic.NewDuration(-1)
+		pid.passivateAfter = -1
 	}
 }
 
diff --git a/cluster/node.go b/cluster/node.go
index 044a935..1b427a2 100644
--- a/cluster/node.go
+++ b/cluster/node.go
@@ -50,10 +50,9 @@ func getHostNode() (*node, error) {
 	}
 	// create the host node
 	return &node{
-		Name:         cfg.Name,
-		Host:         cfg.Host,
-		GossipPort:   cfg.GossipPort,
-		ClusterPort:  cfg.ClusterPort,
-		RemotingPort: cfg.RemotingPort,
+		Name:        cfg.Name,
+		Host:        cfg.Host,
+		GossipPort:  cfg.GossipPort,
+		ClusterPort: cfg.ClusterPort,
 	}, nil
 }


Test patch:
diff --git a/actors/pid_option_test.go b/actors/pid_option_test.go
index 52c9a5e..ada82b0 100644
--- a/actors/pid_option_test.go
+++ b/actors/pid_option_test.go
@@ -6,7 +6,6 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/tochemey/goakt/log"
-	"go.uber.org/atomic"
 )
 
 func TestPIDOptions(t *testing.T) {
@@ -18,17 +17,17 @@ func TestPIDOptions(t *testing.T) {
 		{
 			name:           "WithPassivationAfter",
 			option:         withPassivationAfter(time.Second),
-			expectedConfig: &pid{passivateAfter: atomic.NewDuration(time.Second)},
+			expectedConfig: &pid{passivateAfter: time.Second},
 		},
 		{
 			name:           "WithSendReplyTimeout",
 			option:         withSendReplyTimeout(time.Second),
-			expectedConfig: &pid{sendReplyTimeout: atomic.NewDuration(time.Second)},
+			expectedConfig: &pid{sendReplyTimeout: time.Second},
 		},
 		{
 			name:           "WithInitMaxRetries",
 			option:         withInitMaxRetries(5),
-			expectedConfig: &pid{initMaxRetries: atomic.NewInt32(5)},
+			expectedConfig: &pid{initMaxRetries: 5},
 		},
 		{
 			name:           "WithLogger",
@@ -43,12 +42,12 @@ func TestPIDOptions(t *testing.T) {
 		{
 			name:           "WithShutdownTimeout",
 			option:         withShutdownTimeout(time.Second),
-			expectedConfig: &pid{shutdownTimeout: atomic.NewDuration(time.Second)},
+			expectedConfig: &pid{shutdownTimeout: time.Second},
 		},
 		{
 			name:           "WithPassivationDisabled",
 			option:         withPassivationDisabled(),
-			expectedConfig: &pid{passivateAfter: atomic.NewDuration(-1)},
+			expectedConfig: &pid{passivateAfter: -1},
 		},
 	}
 	for _, tc := range testCases {