Repository: Tochemey/goakt
Pull request #93 (2023-08-06 19:07:46)

feat: add api tests

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/Tochemey/goakt/pull/93?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) Report
Patch coverage: **`77.99%`** and project coverage change: **`+6.71%`** :tada:
> Comparison is base [(`f768409`)](https://app.codecov.io/gh/Tochemey/goakt/commit/f768409a62f81ad0723d843d4b9e4203a0174436?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) 62.94% compared to head [(`5f7bc2f`)](https://app.codecov.io/gh/Tochemey/goakt/pull/93?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) 69.66%.

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main      #93      +/-   ##
==========================================
+ Coverage   62.94%   69.66%   +6.71%     
==========================================
  Files          32       32              
  Lines        2634     2647      +13     
==========================================
+ Hits         1658     1844     +186     
+ Misses        911      720     -191     
- Partials       65       83      +18     
```


| [Files Changed](https://app.codecov.io/gh/Tochemey/goakt/pull/93?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) | Coverage Δ | |
|---|---|---|
| [actors/actor\_system.go](https://app.codecov.io/gh/Tochemey/goakt/pull/93?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene#diff-YWN0b3JzL2FjdG9yX3N5c3RlbS5nbw==) | `66.25% <75.82%> (+16.61%)` | :arrow_up: |
| [actors/option.go](https://app.codecov.io/gh/Tochemey/goakt/pull/93?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene#diff-YWN0b3JzL29wdGlvbi5nbw==) | `91.48% <92.59%> (+0.79%)` | :arrow_up: |

... and [2 files with indirect coverage changes](https://app.codecov.io/gh/Tochemey/goakt/pull/93/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene)


</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/Tochemey/goakt/pull/93?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene).


Patch:
diff --git a/actors/actor_system.go b/actors/actor_system.go
index ba32db9..1621b2c 100644
--- a/actors/actor_system.go
+++ b/actors/actor_system.go
@@ -3,6 +3,7 @@ package actors
 import (
 	"context"
 	"fmt"
+	"net"
 	"net/http"
 	"regexp"
 	"sync"
@@ -92,6 +93,8 @@ type actorSystem struct {
 	// Specifies how long the sender of a receiveContext should wait to receive a reply
 	// when using SendReply. The default value is 5s
 	replyTimeout time.Duration
+	// Specifies the shutdown timeout. The default value is 30s
+	shutdownTimeout time.Duration
 	// Specifies the maximum of retries to attempt when the actor
 	// initialization fails. The default value is 5
 	actorInitMaxRetries int
@@ -138,79 +141,82 @@ func NewActorSystem(name string, opts ...Option) (ActorSystem, error) {
 		return nil, ErrInvalidActorSystemName
 	}
 
-	// the function only gets called one
-	once.Do(func() {
-		system = &actorSystem{
-			actors:              cmp.New[PID](),
-			hasStarted:          atomic.NewBool(false),
-			typesLoader:         NewTypesLoader(),
-			clusterChan:         make(chan *goaktpb.WireActor, 10),
-			name:                name,
-			logger:              log.DefaultLogger,
-			expireActorAfter:    2 * time.Second,
-			replyTimeout:        100 * time.Millisecond,
-			actorInitMaxRetries: 5,
-			supervisorStrategy:  StopDirective,
-			telemetry:           telemetry.New(),
-			remotingEnabled:     atomic.NewBool(false),
-			clusterEnabled:      atomic.NewBool(false),
-			mu:                  sync.Mutex{},
-		}
-		// set the reflection
-		system.reflection = NewReflection(system.typesLoader)
-		// apply the various options
-		for _, opt := range opts {
-			opt.Apply(system)
-		}
-	})
+	if system == nil {
+		// the function only gets called one
+		once.Do(func() {
+			system = &actorSystem{
+				actors:              cmp.New[PID](),
+				hasStarted:          atomic.NewBool(false),
+				typesLoader:         NewTypesLoader(),
+				clusterChan:         make(chan *goaktpb.WireActor, 10),
+				name:                name,
+				logger:              log.DefaultLogger,
+				expireActorAfter:    2 * time.Second,
+				replyTimeout:        100 * time.Millisecond,
+				actorInitMaxRetries: 5,
+				supervisorStrategy:  StopDirective,
+				telemetry:           telemetry.New(),
+				remotingEnabled:     atomic.NewBool(false),
+				clusterEnabled:      atomic.NewBool(false),
+				mu:                  sync.Mutex{},
+				shutdownTimeout:     30 * time.Second,
+			}
+			// set the reflection
+			system.reflection = NewReflection(system.typesLoader)
+			// apply the various options
+			for _, opt := range opts {
+				opt.Apply(system)
+			}
+		})
+	}
 
 	return system, nil
 }
 
 // GetPartition returns the partition where a given actor is located
-func (a *actorSystem) GetPartition(ctx context.Context, actorName string) uint64 {
+func (asys *actorSystem) GetPartition(ctx context.Context, actorName string) uint64 {
 	// add a span context
 	ctx, span := telemetry.SpanContext(ctx, "GetPartition")
 	defer span.End()
 
 	// return zero when the actor system is not in cluster mode
-	if !a.InCluster() {
+	if !asys.InCluster() {
 		// TODO: maybe add a partitioner function
 		return 0
 	}
 
 	// fetch the actor name partition from the cluster
-	return uint64(a.cluster.GetPartition(actorName))
+	return uint64(asys.cluster.GetPartition(actorName))
 }
 
 // InCluster states whether the actor system is running within a cluster of nodes
-func (a *actorSystem) InCluster() bool {
-	return a.clusterEnabled.Load() && a.cluster != nil
+func (asys *actorSystem) InCluster() bool {
+	return asys.clusterEnabled.Load() && asys.cluster != nil
 }
 
 // NumActors returns the total number of active actors in the system
-func (a *actorSystem) NumActors() uint64 {
-	return uint64(a.actors.Count())
+func (asys *actorSystem) NumActors() uint64 {
+	return uint64(asys.actors.Count())
 }
 
 // StartActor creates or returns the instance of a given actor in the system
-func (a *actorSystem) StartActor(ctx context.Context, name string, actor Actor) PID {
+func (asys *actorSystem) StartActor(ctx context.Context, name string, actor Actor) PID {
 	// add a span context
 	ctx, span := telemetry.SpanContext(ctx, "StartActor")
 	defer span.End()
 	// first check whether the actor system has started
-	if !a.hasStarted.Load() {
+	if !asys.hasStarted.Load() {
 		return nil
 	}
 	// set the default actor path assuming we are running locally
-	actorPath := NewPath(name, NewAddress(protocol, a.name, "", -1))
+	actorPath := NewPath(name, NewAddress(protocol, asys.name, "", -1))
 	// set the actor path when the remoting is enabled
-	if a.remotingEnabled.Load() {
+	if asys.remotingEnabled.Load() {
 		// get the path of the given actor
-		actorPath = NewPath(name, NewAddress(protocol, a.name, a.remotingHost, int(a.remotingPort)))
+		actorPath = NewPath(name, NewAddress(protocol, asys.name, asys.remotingHost, int(asys.remotingPort)))
 	}
 	// check whether the given actor already exist in the system or not
-	pid, exist := a.actors.Get(actorPath.String())
+	pid, exist := asys.actors.Get(actorPath.String())
 	// actor already exist no need recreate it.
 	if exist {
 		// check whether the given actor heart beat
@@ -224,22 +230,22 @@ func (a *actorSystem) StartActor(ctx context.Context, name string, actor Actor)
 	pid = newPID(ctx,
 		actorPath,
 		actor,
-		withInitMaxRetries(a.actorInitMaxRetries),
-		withPassivationAfter(a.expireActorAfter),
-		withSendReplyTimeout(a.replyTimeout),
-		withCustomLogger(a.logger),
-		withActorSystem(a),
-		withSupervisorStrategy(a.supervisorStrategy),
-		withTelemetry(a.telemetry))
+		withInitMaxRetries(asys.actorInitMaxRetries),
+		withPassivationAfter(asys.expireActorAfter),
+		withSendReplyTimeout(asys.replyTimeout),
+		withCustomLogger(asys.logger),
+		withActorSystem(asys),
+		withSupervisorStrategy(asys.supervisorStrategy),
+		withTelemetry(asys.telemetry))
 
 	// add the given actor to the actor map
-	a.actors.Set(actorPath.String(), pid)
+	asys.actors.Set(actorPath.String(), pid)
 
 	// let us register the actor
-	a.typesLoader.Register(name, actor)
+	asys.typesLoader.Register(name, actor)
 
 	// when cluster is enabled replicate the actor metadata across the cluster
-	if a.clusterEnabled.Load() {
+	if asys.clusterEnabled.Load() {
 		// TODO: revisit the encoding the actor
 		// encode the actor
 		//actorType := reflect.TypeOf(actor)
@@ -257,7 +263,7 @@ func (a *actorSystem) StartActor(ctx context.Context, name string, actor Actor)
 			ActorPath:    actorPath.String(),
 		}
 		// send it to the cluster channel
-		a.clusterChan <- wiredActor
+		asys.clusterChan <- wiredActor
 	}
 
 	// return the actor ref
@@ -265,23 +271,23 @@ func (a *actorSystem) StartActor(ctx context.Context, name string, actor Actor)
 }
 
 // StopActor stops a given actor in the system
-func (a *actorSystem) StopActor(ctx context.Context, name string) error {
+func (asys *actorSystem) StopActor(ctx context.Context, name string) error {
 	// add a span context
 	ctx, span := telemetry.SpanContext(ctx, "StopActor")
 	defer span.End()
 	// first check whether the actor system has started
-	if !a.hasStarted.Load() {
+	if !asys.hasStarted.Load() {
 		return errors.New("actor system has not started yet")
 	}
 	// set the default actor path assuming we are running locally
-	actorPath := NewPath(name, NewAddress(protocol, a.name, "", -1))
+	actorPath := NewPath(name, NewAddress(protocol, asys.name, "", -1))
 	// set the actor path with the remoting is enabled
-	if a.remotingEnabled.Load() {
+	if asys.remotingEnabled.Load() {
 		// get the path of the given actor
-		actorPath = NewPath(name, NewAddress(protocol, a.name, a.remotingHost, int(a.remotingPort)))
+		actorPath = NewPath(name, NewAddress(protocol, asys.name, asys.remotingHost, int(asys.remotingPort)))
 	}
 	// check whether the given actor already exist in the system or not
-	pid, exist := a.actors.Get(actorPath.String())
+	pid, exist := asys.actors.Get(actorPath.String())
 	// actor is found.
 	if exist {
 		// stop the given actor
@@ -291,23 +297,23 @@ func (a *actorSystem) StopActor(ctx context.Context, name string) error {
 }
 
 // RestartActor restarts a given actor in the system
-func (a *actorSystem) RestartActor(ctx context.Context, name string) (PID, error) {
+func (asys *actorSystem) RestartActor(ctx context.Context, name string) (PID, error) {
 	// add a span context
 	ctx, span := telemetry.SpanContext(ctx, "RestartActor")
 	defer span.End()
 	// first check whether the actor system has started
-	if !a.hasStarted.Load() {
+	if !asys.hasStarted.Load() {
 		return nil, errors.New("actor system has not started yet")
 	}
 	// set the default actor path assuming we are running locally
-	actorPath := NewPath(name, NewAddress(protocol, a.name, "", -1))
+	actorPath := NewPath(name, NewAddress(protocol, asys.name, "", -1))
 	// set the actor path with the remoting is enabled
-	if a.remotingEnabled.Load() {
+	if asys.remotingEnabled.Load() {
 		// get the path of the given actor
-		actorPath = NewPath(name, NewAddress(protocol, a.name, a.remotingHost, int(a.remotingPort)))
+		actorPath = NewPath(name, NewAddress(protocol, asys.name, asys.remotingHost, int(asys.remotingPort)))
 	}
 	// check whether the given actor already exist in the system or not
-	pid, exist := a.actors.Get(actorPath.String())
+	pid, exist := asys.actors.Get(actorPath.String())
 	// actor is found.
 	if exist {
 		// restart the given actor
@@ -321,23 +327,23 @@ func (a *actorSystem) RestartActor(ctx context.Context, name string) (PID, error
 }
 
 // Name returns the actor system name
-func (a *actorSystem) Name() string {
+func (asys *actorSystem) Name() string {
 	// acquire the lock
-	a.mu.Lock()
+	asys.mu.Lock()
 	// release the lock
-	defer a.mu.Unlock()
-	return a.name
+	defer asys.mu.Unlock()
+	return asys.name
 }
 
 // Actors returns the list of Actors that are alive in the actor system
-func (a *actorSystem) Actors() []PID {
+func (asys *actorSystem) Actors() []PID {
 	// acquire the lock
-	a.mu.Lock()
+	asys.mu.Lock()
 	// release the lock
-	defer a.mu.Unlock()
+	defer asys.mu.Unlock()
 
 	// get the actors from the actor map
-	items := a.actors.Items()
+	items := asys.actors.Items()
 	var refs []PID
 	for _, actorRef := range items {
 		refs = append(refs, actorRef)
@@ -348,17 +354,17 @@ func (a *actorSystem) Actors() []PID {
 
 // GetLocalActor returns the reference of a local actor.
 // A local actor is an actor that reside on the same node where the given actor system is running
-func (a *actorSystem) GetLocalActor(ctx context.Context, actorName string) (PID, error) {
+func (asys *actorSystem) GetLocalActor(ctx context.Context, actorName string) (PID, error) {
 	// acquire the lock
-	a.mu.Lock()
+	asys.mu.Lock()
 	// release the lock
-	defer a.mu.Unlock()
+	defer asys.mu.Unlock()
 
 	// add a span context
 	ctx, span := telemetry.SpanContext(ctx, "GetLocalActor")
 	defer span.End()
 	// first let us do a local lookup
-	items := a.actors.Items()
+	items := asys.actors.Items()
 	// iterate the local actors storage
 	for _, actorRef := range items {
 		if actorRef.ActorPath().Name() == actorName {
@@ -366,31 +372,31 @@ func (a *actorSystem) GetLocalActor(ctx context.Context, actorName string) (PID,
 		}
 	}
 	// add a logger
-	a.logger.Infof("actor=%s not found", actorName)
+	asys.logger.Infof("actor=%s not found", actorName)
 	return nil, ErrActorNotFound
 }
 
 // GetRemoteActor returns the address of a remote actor when cluster is enabled
-func (a *actorSystem) GetRemoteActor(ctx context.Context, actorName string) (addr *pb.Address, err error) {
+func (asys *actorSystem) GetRemoteActor(ctx context.Context, actorName string) (addr *pb.Address, err error) {
 	// acquire the lock
-	a.mu.Lock()
+	asys.mu.Lock()
 	// release the lock
-	defer a.mu.Unlock()
+	defer asys.mu.Unlock()
 
 	// add a span context
 	ctx, span := telemetry.SpanContext(ctx, "GetRemoteActor")
 	defer span.End()
 	// check whether cluster is enabled or not
-	if a.cluster == nil {
+	if asys.cluster == nil {
 		return nil, errors.New("cluster is not enabled")
 	}
 
 	// let us locate the actor in the cluster
-	wireActor, err := a.cluster.GetActor(ctx, actorName)
+	wireActor, err := asys.cluster.GetActor(ctx, actorName)
 	// handle the eventual error
 	if err != nil {
 		if errors.Is(err, cluster.ErrActorNotFound) {
-			a.logger.Infof("actor=%s not found", actorName)
+			asys.logger.Infof("actor=%s not found", actorName)
 			return nil, ErrActorNotFound
 		}
 		return nil, errors.Wrapf(err, "failed to fetch remote actor=%s", actorName)
@@ -401,69 +407,73 @@ func (a *actorSystem) GetRemoteActor(ctx context.Context, actorName string) (add
 }
 
 // Start starts the actor system
-func (a *actorSystem) Start(ctx context.Context) error {
+func (asys *actorSystem) Start(ctx context.Context) error {
 	// add a span context
 	ctx, span := telemetry.SpanContext(ctx, "Start")
 	defer span.End()
 	// set the has started to true
-	a.hasStarted.Store(true)
+	asys.hasStarted.Store(true)
 
 	// start remoting when remoting is enabled
-	if a.remotingEnabled.Load() {
-		a.enableRemoting(ctx)
+	if asys.remotingEnabled.Load() {
+		asys.enableRemoting(ctx)
 	}
 
 	// enable clustering when it is enabled
-	if a.clusterEnabled.Load() {
-		a.enableClustering(ctx)
+	if asys.clusterEnabled.Load() {
+		asys.enableClustering(ctx)
 	}
 
 	// start the metrics service
 	// register metrics. However, we don't panic when we fail to register
 	// we just log it for now
 	// TODO decide what to do when we fail to register the metrics or export the metrics registration as public
-	if err := a.registerMetrics(); err != nil {
-		a.logger.Error(errors.Wrapf(err, "failed to register actorSystem=%s metrics", a.name))
+	if err := asys.registerMetrics(); err != nil {
+		asys.logger.Error(errors.Wrapf(err, "failed to register actorSystem=%s metrics", asys.name))
 	}
-	a.logger.Infof("%s started..:)", a.name)
+	asys.logger.Infof("%s started..:)", asys.name)
 	return nil
 }
 
 // Stop stops the actor system
-func (a *actorSystem) Stop(ctx context.Context) error {
+func (asys *actorSystem) Stop(ctx context.Context) error {
+	// create a cancellation context to gracefully shutdown
+	ctx, cancel := context.WithTimeout(ctx, asys.shutdownTimeout)
+	defer cancel()
+
 	// add a span context
 	ctx, span := telemetry.SpanContext(ctx, "Stop")
 	defer span.End()
-	a.logger.Infof("%s is shutting down..:)", a.name)
+	asys.logger.Infof("%s is shutting down..:)", asys.name)
 
 	// stop the remoting server
-	if a.remotingEnabled.Load() {
+	if asys.remotingEnabled.Load() {
 		// stop the server
-		if err := a.remotingServer.Close(); err != nil {
+		if err := asys.remotingServer.Shutdown(ctx); err != nil {
 			return err
 		}
 	}
 
 	// stop the cluster service
-	if a.clusterEnabled.Load() {
+	if asys.clusterEnabled.Load() {
 		// stop the cluster service
-		if err := a.cluster.Stop(ctx); err != nil {
+		if err := asys.cluster.Stop(ctx); err != nil {
 			return err
 		}
 		// stop broadcasting cluster messages
-		close(a.clusterChan)
+		close(asys.clusterChan)
 	}
 
 	// short-circuit the shutdown process when there are no online actors
-	if len(a.Actors()) == 0 {
-		a.logger.Info("No online actors to shutdown. Shutting down successfully done")
+	if len(asys.Actors()) == 0 {
+		asys.logger.Info("No online actors to shutdown. Shutting down successfully done")
 		return nil
 	}
 
 	// stop all the actors
-	for _, actor := range a.Actors() {
+	for _, actor := range asys.Actors() {
 		// remove the actor the map and shut it down
-		a.actors.Remove(actor.ActorPath().String())
+		asys.actors.Remove(actor.ActorPath().String())
 		// only shutdown live actors
 		if actor.IsOnline() {
 			if err := actor.Shutdown(ctx); err != nil {
@@ -473,30 +483,30 @@ func (a *actorSystem) Stop(ctx context.Context) error {
 	}
 
 	// reset the actor system
-	a.reset()
+	asys.reset()
 
 	return nil
 }
 
 // RemoteLookup for an actor on a remote host.
-func (a *actorSystem) RemoteLookup(ctx context.Context, request *connect.Request[goaktpb.RemoteLookupRequest]) (*connect.Response[goaktpb.RemoteLookupResponse], error) {
+func (asys *actorSystem) RemoteLookup(ctx context.Context, request *connect.Request[goaktpb.RemoteLookupRequest]) (*connect.Response[goaktpb.RemoteLookupResponse], error) {
 	// add a span context
 	ctx, span := telemetry.SpanContext(ctx, "RemoteLookup")
 	defer span.End()
 
 	// get a context log
-	logger := a.logger
+	logger := asys.logger
 
 	// first let us make a copy of the incoming request
 	reqCopy := request.Msg
 
 	// set the actor path with the remoting is enabled
-	if !a.remotingEnabled.Load() {
+	if !asys.remotingEnabled.Load() {
 		return nil, connect.NewError(connect.CodeFailedPrecondition, ErrRemotingNotEnabled)
 	}
 
 	// get the remoting server address
-	nodeAddr := fmt.Sprintf("%s:%d", a.remotingHost, a.remotingPort)
+	nodeAddr := fmt.Sprintf("%s:%d", asys.remotingHost, asys.remotingPort)
 
 	// let us validate the host and port
 	hostAndPort := fmt.Sprintf("%s:%d", reqCopy.GetHost(), reqCopy.GetPort())
@@ -509,10 +519,10 @@ func (a *actorSystem) RemoteLookup(ctx context.Context, request *connect.Request
 
 	// construct the actor address
 	name := reqCopy.GetName()
-	actorPath := NewPath(name, NewAddress(protocol, a.Name(), reqCopy.GetHost(), int(reqCopy.GetPort())))
+	actorPath := NewPath(name, NewAddress(protocol, asys.Name(), reqCopy.GetHost(), int(reqCopy.GetPort())))
 	// start or get the PID of the actor
 	// check whether the given actor already exist in the system or not
-	pid, exist := a.actors.Get(actorPath.String())
+	pid, exist := asys.actors.Get(actorPath.String())
 	// return an error when the remote address is not found
 	if !exist {
 		// log the error
@@ -529,23 +539,23 @@ func (a *actorSystem) RemoteLookup(ctx context.Context, request *connect.Request
 // RemoteAsk is used to send a message to an actor remotely and expect a response
 // immediately. With this type of message the receiver cannot communicate back to Sender
 // except reply the message with a response. This one-way communication
-func (a *actorSystem) RemoteAsk(ctx context.Context, request *connect.Request[goaktpb.RemoteAskRequest]) (*connect.Response[goaktpb.RemoteAskResponse], error) {
+func (asys *actorSystem) RemoteAsk(ctx context.Context, request *connect.Request[goaktpb.RemoteAskRequest]) (*connect.Response[goaktpb.RemoteAskResponse], error) {
 	// add a span context
 	ctx, span := telemetry.SpanContext(ctx, "RemoteAsk")
 	defer span.End()
 
 	// get a context log
-	logger := a.logger
+	logger := asys.logger
 	// first let us make a copy of the incoming request
 	reqCopy := request.Msg
 
 	// set the actor path with the remoting is enabled
-	if !a.remotingEnabled.Load() {
+	if !asys.remotingEnabled.Load() {
 		return nil, connect.NewError(connect.CodeFailedPrecondition, ErrRemotingNotEnabled)
 	}
 
 	// get the remoting server address
-	nodeAddr := fmt.Sprintf("%s:%d", a.remotingHost, a.remotingPort)
+	nodeAddr := fmt.Sprintf("%s:%d", asys.remotingHost, asys.remotingPort)
 
 	// let us validate the host and port
 	hostAndPort := fmt.Sprintf("%s:%d", reqCopy.GetRemoteMessage().GetReceiver().GetHost(), reqCopy.GetRemoteMessage().GetReceiver().GetPort())
@@ -558,11 +568,11 @@ func (a *actorSystem) RemoteAsk(ctx context.Context, request *connect.Request[go
 
 	// construct the actor address
 	name := reqCopy.GetRemoteMessage().GetReceiver().GetName()
-	actorPath := NewPath(name, NewAddress(protocol, a.name, a.remotingHost, int(a.remotingPort)))
+	actorPath := NewPath(name, NewAddress(protocol, asys.name, asys.remotingHost, int(asys.remotingPort)))
 
 	// start or get the PID of the actor
 	// check whether the given actor already exist in the system or not
-	pid, exist := a.actors.Get(actorPath.String())
+	pid, exist := asys.actors.Get(actorPath.String())
 	// return an error when the remote address is not found
 	if !exist {
 		// log the error
@@ -577,7 +587,7 @@ func (a *actorSystem) RemoteAsk(ctx context.Context, request *connect.Request[go
 	}
 
 	// send the message to actor
-	reply, err := a.handleRemoteAsk(ctx, pid, reqCopy.GetRemoteMessage())
+	reply, err := asys.handleRemoteAsk(ctx, pid, reqCopy.GetRemoteMessage())
 	// handle the error
 	if err != nil {
 		logger.Error(ErrRemoteSendFailure(err).Error())
@@ -591,25 +601,25 @@ func (a *actorSystem) RemoteAsk(ctx context.Context, request *connect.Request[go
 // RemoteTell is used to send a message to an actor remotely by another actor
 // This is the only way remote actors can interact with each other. The actor on the
 // other line can reply to the sender by using the Sender in the message
-func (a *actorSystem) RemoteTell(ctx context.Context, request *connect.Request[goaktpb.RemoteTellRequest]) (*connect.Response[goaktpb.RemoteTellResponse], error) {
+func (asys *actorSystem) RemoteTell(ctx context.Context, request *connect.Request[goaktpb.RemoteTellRequest]) (*connect.Response[goaktpb.RemoteTellResponse], error) {
 	// add a span context
 	ctx, span := telemetry.SpanContext(ctx, "RemoteTell")
 	defer span.End()
 
 	// get a context log
-	logger := a.logger
+	logger := asys.logger
 	// first let us make a copy of the incoming request
 	reqCopy := request.Msg
 
 	receiver := reqCopy.GetRemoteMessage().GetReceiver()
 
 	// set the actor path with the remoting is enabled
-	if !a.remotingEnabled.Load() {
+	if !asys.remotingEnabled.Load() {
 		return nil, connect.NewError(connect.CodeFailedPrecondition, ErrRemotingNotEnabled)
 	}
 
 	// get the remoting server address
-	nodeAddr := fmt.Sprintf("%s:%d", a.remotingHost, a.remotingPort)
+	nodeAddr := fmt.Sprintf("%s:%d", asys.remotingHost, asys.remotingPort)
 
 	// let us validate the host and port
 	hostAndPort := fmt.Sprintf("%s:%d", receiver.GetHost(), receiver.GetPort())
@@ -625,12 +635,12 @@ func (a *actorSystem) RemoteTell(ctx context.Context, request *connect.Request[g
 		receiver.GetName(),
 		NewAddress(
 			protocol,
-			a.Name(),
+			asys.Name(),
 			receiver.GetHost(),
 			int(receiver.GetPort())))
 	// start or get the PID of the actor
 	// check whether the given actor already exist in the system or not
-	pid, exist := a.actors.Get(actorPath.String())
+	pid, exist := asys.actors.Get(actorPath.String())
 	// return an error when the remote address is not found
 	if !exist {
 		// log the error
@@ -645,7 +655,7 @@ func (a *actorSystem) RemoteTell(ctx context.Context, request *connect.Request[g
 	}
 
 	// send the message to actor
-	if err := a.handleRemoteTell(ctx, pid, reqCopy.GetRemoteMessage()); err != nil {
+	if err := asys.handleRemoteTell(ctx, pid, reqCopy.GetRemoteMessage()); err != nil {
 		logger.Error(ErrRemoteSendFailure(err))
 		return nil, ErrRemoteSendFailure(err)
 	}
@@ -653,9 +663,9 @@ func (a *actorSystem) RemoteTell(ctx context.Context, request *connect.Request[g
 }
 
 // registerMetrics register the PID metrics with OTel instrumentation.
-func (a *actorSystem) registerMetrics() error {
+func (asys *actorSystem) registerMetrics() error {
 	// grab the OTel meter
-	meter := a.telemetry.Meter
+	meter := asys.telemetry.Meter
 	// create an instance of the ActorMetrics
 	metrics, err := telemetry.NewSystemMetrics(meter)
 	// handle the error
@@ -665,12 +675,12 @@ func (a *actorSystem) registerMetrics() error {
 
 	// define the common labels
 	labels := []attribute.KeyValue{
-		attribute.String("actor.system", a.Name()),
+		attribute.String("actor.system", asys.Name()),
 	}
 
 	// register the metrics
 	_, err = meter.RegisterCallback(func(ctx context.Context, observer metric.Observer) error {
-		observer.ObserveInt64(metrics.ActorSystemActorsCount, int64(a.NumActors()), metric.WithAttributes(labels...))
+		observer.ObserveInt64(metrics.ActorSystemActorsCount, int64(asys.NumActors()), metric.WithAttributes(labels...))
 		return nil
 	}, metrics.ActorSystemActorsCount)
 
@@ -679,15 +689,15 @@ func (a *actorSystem) registerMetrics() error {
 
 // handleRemoteAsk handles a synchronous message to another actor and expect a response.
 // This block until a response is received or timed out.
-func (a *actorSystem) handleRemoteAsk(ctx context.Context, to PID, message proto.Message) (response proto.Message, err error) {
+func (asys *actorSystem) handleRemoteAsk(ctx context.Context, to PID, message proto.Message) (response proto.Message, err error) {
 	// add a span context
 	ctx, span := telemetry.SpanContext(ctx, "handleRemoteAsk")
 	defer span.End()
-	return Ask(ctx, to, message, a.replyTimeout)
+	return Ask(ctx, to, message, asys.replyTimeout)
 }
 
 // handleRemoteTell handles an asynchronous message to an actor
-func (a *actorSystem) handleRemoteTell(ctx context.Context, to PID, message proto.Message) error {
+func (asys *actorSystem) handleRemoteTell(ctx context.Context, to PID, message proto.Message) error {
 	// add a span context
 	ctx, span := telemetry.SpanContext(ctx, "handleRemoteTell")
 	defer span.End()
@@ -696,26 +706,26 @@ func (a *actorSystem) handleRemoteTell(ctx context.Context, to PID, message prot
 
 // enableClustering enables clustering. When clustering is enabled remoting is also enabled to facilitate remote
 // communication
-func (a *actorSystem) enableClustering(ctx context.Context) {
+func (asys *actorSystem) enableClustering(ctx context.Context) {
 	// add some logging information
-	a.logger.Info("enabling clustering...")
+	asys.logger.Info("enabling clustering...")
 
 	// create an instance of the cluster service and start it
-	cluster, err := cluster.New(a.Name(),
-		a.serviceDiscovery,
-		cluster.WithLogger(a.logger),
-		cluster.WithPartitionsCount(a.partitionsCount),
+	cluster, err := cluster.New(asys.Name(),
+		asys.serviceDiscovery,
+		cluster.WithLogger(asys.logger),
+		cluster.WithPartitionsCount(asys.partitionsCount),
 	)
 	// handle the error
 	if err != nil {
-		a.logger.Panic(errors.Wrap(err, "failed to initialize cluster engine"))
+		asys.logger.Panic(errors.Wrap(err, "failed to initialize cluster engine"))
 	}
 
 	// add some logging information
-	a.logger.Info("starting cluster engine...")
+	asys.logger.Info("starting cluster engine...")
 	// start the cluster service
 	if err := cluster.Start(ctx); err != nil {
-		a.logger.Panic(errors.Wrap(err, "failed to start cluster engine"))
+		asys.logger.Panic(errors.Wrap(err, "failed to start cluster engine"))
 	}
 
 	// create the bootstrap channel
@@ -728,33 +738,33 @@ func (a *actorSystem) enableClustering(ctx context.Context) {
 	<-bootstrapChan
 	timer.Stop()
 
-	a.logger.Info("cluster engine successfully started...")
+	asys.logger.Info("cluster engine successfully started...")
 
 	// acquire the lock
-	a.mu.Lock()
+	asys.mu.Lock()
 	// set the cluster field
-	a.cluster = cluster
+	asys.cluster = cluster
 	// release the lock
 	// set the remoting host and port
-	a.remotingHost = cluster.NodeHost()
-	a.remotingPort = int32(cluster.NodeRemotingPort())
+	asys.remotingHost = cluster.NodeHost()
+	asys.remotingPort = int32(cluster.NodeRemotingPort())
 	// release the lock
-	a.mu.Unlock()
+	asys.mu.Unlock()
 
 	// let us enable remoting as well if not yet enabled
-	if !a.remotingEnabled.Load() {
-		a.enableRemoting(ctx)
+	if !asys.remotingEnabled.Load() {
+		asys.enableRemoting(ctx)
 	}
 	// start broadcasting cluster message
-	go a.broadcast(ctx)
+	go asys.broadcast(ctx)
 	// add some logging
-	a.logger.Info("clustering enabled...:)")
+	asys.logger.Info("clustering enabled...:)")
 }
 
 // enableRemoting enables the remoting service to handle remote messaging
-func (a *actorSystem) enableRemoting(context.Context) {
+func (asys *actorSystem) enableRemoting(ctx context.Context) {
 	// add some logging information
-	a.logger.Info("enabling remoting...")
+	asys.logger.Info("enabling remoting...")
 	// create a function to handle the observability
 	interceptor := func(tp trace.TracerProvider, mp metric.MeterProvider) connect.Interceptor {
 		return otelconnect.NewInterceptor(
@@ -767,12 +777,12 @@ func (a *actorSystem) enableRemoting(context.Context) {
 	mux := http.NewServeMux()
 	// create the resource and handler
 	path, handler := goaktv1connect.NewRemoteMessagingServiceHandler(
-		a,
-		connect.WithInterceptors(interceptor(a.telemetry.TracerProvider, a.telemetry.MeterProvider)),
+		asys,
+		connect.WithInterceptors(interceptor(asys.telemetry.TracerProvider, asys.telemetry.MeterProvider)),
 	)
 	mux.Handle(path, handler)
 	// create the address
-	serverAddr := fmt.Sprintf("%s:%d", a.remotingHost, a.remotingPort)
+	serverAddr := fmt.Sprintf("%s:%d", asys.remotingHost, asys.remotingPort)
 
 	// create a http service instance
 	// TODO revisit the timeouts
@@ -797,6 +807,10 @@ func (a *actorSystem) enableRemoting(context.Context) {
 		Handler: h2c.NewHandler(mux, &http2.Server{
 			IdleTimeout: 1200 * time.Second,
 		}),
+		// Set the base context to incoming context of the system
+		BaseContext: func(listener net.Listener) context.Context {
+			return ctx
+		},
 	}
 
 	// listen and service requests
@@ -804,45 +818,46 @@ func (a *actorSystem) enableRemoting(context.Context) {
 		if err := server.ListenAndServe(); err != nil {
 			// check the error type
 			if !errors.Is(err, http.ErrServerClosed) {
-				a.logger.Panic(errors.Wrap(err, "failed to start remoting service"))
+				asys.logger.Panic(errors.Wrap(err, "failed to start remoting service"))
 			}
 		}
 	}()
 
 	// acquire the lock
-	a.mu.Lock()
+	asys.mu.Lock()
 	// set the server
-	a.remotingServer = server
+	asys.remotingServer = server
 	// release the lock
-	a.mu.Unlock()
+	asys.mu.Unlock()
 
 	// add some logging information
-	a.logger.Info("remoting enabled...:)")
+	asys.logger.Info("remoting enabled...:)")
 }
 
 // reset the actor system
-func (a *actorSystem) reset() {
-	// void the settings
-	a.hasStarted = atomic.NewBool(false)
-	a.remotingEnabled = atomic.NewBool(false)
-	a.clusterEnabled = atomic.NewBool(false)
-	a.telemetry = nil
-	a.actors = cmp.New[PID]()
-	a.name = ""
-	a.logger = nil
-	a.remotingServer = nil
+func (asys *actorSystem) reset() {
+	asys.hasStarted = atomic.NewBool(false)
+	asys.remotingEnabled = atomic.NewBool(false)
+	asys.clusterEnabled = atomic.NewBool(false)
+	asys.telemetry = nil
+	asys.actors = cmp.New[PID]()
+	asys.name = ""
+	asys.logger = nil
+	asys.remotingServer = nil
+	// set the global nil
+	system = nil
 	once.Reset()
 }
 
 // broadcast publishes newly created actor into the cluster when cluster is enabled
-func (a *actorSystem) broadcast(ctx context.Context) {
+func (asys *actorSystem) broadcast(ctx context.Context) {
 	// iterate over the channel
-	for wireActor := range a.clusterChan {
+	for wireActor := range asys.clusterChan {
 		// making sure the cluster is still enabled
-		if a.cluster != nil {
+		if asys.cluster != nil {
 			// broadcast the message on the cluster
-			if err := a.cluster.PutActor(ctx, wireActor); err != nil {
-				a.logger.Error(err.Error())
+			if err := asys.cluster.PutActor(ctx, wireActor); err != nil {
+				asys.logger.Error(err.Error())
 				// TODO: stop or continue
 				return
 			}
diff --git a/actors/option.go b/actors/option.go
index bf4328f..792f0eb 100644
--- a/actors/option.go
+++ b/actors/option.go
@@ -33,70 +33,77 @@ func (f OptionFunc) Apply(c *actorSystem) {
 // WithExpireActorAfter sets the actor expiry duration.
 // After such duration an idle actor will be expired and removed from the actor system
 func WithExpireActorAfter(duration time.Duration) Option {
-	return OptionFunc(func(sys *actorSystem) {
-		sys.expireActorAfter = duration
+	return OptionFunc(func(a *actorSystem) {
+		a.expireActorAfter = duration
 	})
 }
 
 // WithLogger sets the actor system custom log
 func WithLogger(logger log.Logger) Option {
-	return OptionFunc(func(sys *actorSystem) {
-		sys.logger = logger
+	return OptionFunc(func(a *actorSystem) {
+		a.logger = logger
 	})
 }
 
 // WithReplyTimeout sets how long in seconds an actor should reply a command
 // in a receive-reply pattern
 func WithReplyTimeout(timeout time.Duration) Option {
-	return OptionFunc(func(sys *actorSystem) {
-		sys.replyTimeout = timeout
+	return OptionFunc(func(a *actorSystem) {
+		a.replyTimeout = timeout
 	})
 }
 
 // WithActorInitMaxRetries sets the number of times to retry an actor init process
 func WithActorInitMaxRetries(max int) Option {
-	return OptionFunc(func(sys *actorSystem) {
-		sys.actorInitMaxRetries = max
+	return OptionFunc(func(a *actorSystem) {
+		a.actorInitMaxRetries = max
 	})
 }
 
 // WithPassivationDisabled disable the passivation mode
 func WithPassivationDisabled() Option {
-	return OptionFunc(func(sys *actorSystem) {
-		sys.expireActorAfter = -1
+	return OptionFunc(func(a *actorSystem) {
+		a.expireActorAfter = -1
 	})
 }
 
 // WithSupervisorStrategy sets the supervisor strategy
 func WithSupervisorStrategy(strategy StrategyDirective) Option {
-	return OptionFunc(func(sys *actorSystem) {
-		sys.supervisorStrategy = strategy
+	return OptionFunc(func(a *actorSystem) {
+		a.supervisorStrategy = strategy
 	})
 }
 
 // WithTelemetry sets the custom telemetry
 func WithTelemetry(telemetry *telemetry.Telemetry) Option {
-	return OptionFunc(func(sys *actorSystem) {
-		sys.telemetry = telemetry
+	return OptionFunc(func(a *actorSystem) {
+		a.telemetry = telemetry
 	})
 }
 
 // WithRemoting enables remoting on the actor system
 func WithRemoting(host string, port int32) Option {
-	return OptionFunc(func(sys *actorSystem) {
-		sys.remotingEnabled = atomic.NewBool(true)
-		sys.remotingPort = port
-		sys.remotingHost = host
+	return OptionFunc(func(a *actorSystem) {
+		a.remotingEnabled = atomic.NewBool(true)
+		a.remotingPort = port
+		a.remotingHost = host
 	})
 }
 
 // WithClustering enables clustering on the actor system. This enables remoting on the actor system as well
 // and set the remotingHost to the cluster node host when the cluster is fully enabled.
 func WithClustering(serviceDiscovery *discovery.ServiceDiscovery, partitionCount uint64) Option {
-	return OptionFunc(func(sys *actorSystem) {
-		sys.clusterEnabled = atomic.NewBool(true)
-		sys.remotingEnabled = atomic.NewBool(true)
-		sys.partitionsCount = partitionCount
-		sys.serviceDiscovery = serviceDiscovery
+	return OptionFunc(func(a *actorSystem) {
+		a.clusterEnabled = atomic.NewBool(true)
+		a.remotingEnabled = atomic.NewBool(true)
+		a.partitionsCount = partitionCount
+		a.serviceDiscovery = serviceDiscovery
+	})
+}
+
+// WithShutdownTimeout sets the shutdown timeout
+func WithShutdownTimeout(timeout time.Duration) Option {
+	return OptionFunc(func(a *actorSystem) {
+		a.shutdownTimeout = timeout
 	})
 }
diff --git a/readme.md b/readme.md
index 1895060..d4e3911 100644
--- a/readme.md
+++ b/readme.md
@@ -6,12 +6,12 @@
 [![GitHub release (with filter)](https://img.shields.io/github/v/release/tochemey/goakt)](https://github.com/Tochemey/goakt/releases)
 
 
-Minimal [Go](https://go.dev/) actor framework with goodies to build reactive and distributed system in golang using
+Distributed [Go](https://go.dev/) actor framework to build reactive and distributed system in golang using
 _**protocol buffers as actor messages**_.
 
 GoAkt is highly scalable and available when running in cluster mode. It comes with the necessary features require to
 build a distributed actor-based system without
-sacrificing performance and reliability.
+sacrificing performance and reliability. With GoAkt, you can instantly create a fast, scalable, distributed system across a cluster of computers.
 
 If you are not familiar with the actor model, the blog post from Brian
 Storti [here](https://www.brianstorti.com/the-actor-model/) is an excellent and short introduction to the actor model.


Test patch:
diff --git a/actors/api_test.go b/actors/api_test.go
new file mode 100644
index 0000000..0ada8c0
--- /dev/null
+++ b/actors/api_test.go
@@ -0,0 +1,161 @@
+package actors
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/tochemey/goakt/log"
+	testpb "github.com/tochemey/goakt/test/data/pb/v1"
+	"github.com/travisjeffery/go-dynaport"
+	"google.golang.org/protobuf/proto"
+)
+
+func TestAsk(t *testing.T) {
+	// create the context
+	ctx := context.TODO()
+	// define the logger to use
+	logger := log.New(log.DebugLevel, os.Stdout)
+	// create the actor system
+	sys, err := NewActorSystem("test",
+		WithLogger(logger),
+		WithPassivationDisabled())
+	// assert there are no error
+	require.NoError(t, err)
+
+	// start the actor system
+	err = sys.Start(ctx)
+	assert.NoError(t, err)
+
+	// create a test actor
+	actorName := "test"
+	actor := NewTestActor()
+	actorRef := sys.StartActor(ctx, actorName, actor)
+	assert.NotNil(t, actorRef)
+
+	// create a message to send to the test actor
+	message := new(testpb.TestReply)
+	// send the message to the actor
+	reply, err := Ask(ctx, actorRef, message, receivingTimeout)
+	// perform some assertions
+	require.NoError(t, err)
+	assert.NotNil(t, reply)
+	expected := &testpb.Reply{Content: "received message"}
+	assert.True(t, proto.Equal(expected, reply))
+
+	// stop the actor after some time
+	ctx, cancel := context.WithTimeout(ctx, time.Second)
+	defer cancel()
+
+	err = sys.Stop(ctx)
+}
+
+func TestTell(t *testing.T) {
+	// create the context
+	ctx := context.TODO()
+	// define the logger to use
+	logger := log.New(log.DebugLevel, os.Stdout)
+	// create the actor system
+	sys, err := NewActorSystem("test",
+		WithLogger(logger),
+		WithPassivationDisabled())
+	// assert there are no error
+	require.NoError(t, err)
+
+	// start the actor system
+	err = sys.Start(ctx)
+	assert.NoError(t, err)
+
+	// create a test actor
+	actorName := "test"
+	actor := NewTestActor()
+	actorRef := sys.StartActor(ctx, actorName, actor)
+	assert.NotNil(t, actorRef)
+
+	// create a message to send to the test actor
+	message := new(testpb.TestSend)
+	// send the message to the actor
+	err = Tell(ctx, actorRef, message)
+	// perform some assertions
+	require.NoError(t, err)
+	count := 1
+	assert.EqualValues(t, count, actorRef.ReceivedCount(ctx))
+
+	// stop the actor after some time
+	ctx, cancel := context.WithTimeout(ctx, time.Second)
+	defer cancel()
+
+	err = sys.Stop(ctx)
+}
+
+func TestRemoteMessaging(t *testing.T) {
+	// create the context
+	ctx := context.TODO()
+	// define the logger to use
+	logger := log.New(log.DebugLevel, os.Stdout)
+	// generate the remoting port
+	nodePorts := dynaport.Get(1)
+	remotingPort := nodePorts[0]
+	host := "127.0.0.1"
+
+	// create the actor system
+	sys, err := NewActorSystem("test",
+		WithLogger(logger),
+		WithPassivationDisabled(),
+		WithRemoting(host, int32(remotingPort)),
+	)
+	// assert there are no error
+	require.NoError(t, err)
+
+	// start the actor system
+	err = sys.Start(ctx)
+	assert.NoError(t, err)
+
+	// create a test actor
+	actorName := "test"
+	actor := NewTestActor()
+	actorRef := sys.StartActor(ctx, actorName, actor)
+	assert.NotNil(t, actorRef)
+
+	// get the address of the actor
+	addr, err := RemoteLookup(ctx, host, remotingPort, actorName)
+	require.NoError(t, err)
+
+	var message proto.Message
+	// create a message to send to the test actor
+	message = new(testpb.TestReply)
+	// send the message to the actor
+	reply, err := RemoteAsk(ctx, addr, message)
+	// perform some assertions
+	require.NoError(t, err)
+	require.NotNil(t, reply)
+	require.True(t, reply.MessageIs(new(testpb.Reply)))
+
+	actual := new(testpb.Reply)
+	err = reply.UnmarshalTo(actual)
+	require.NoError(t, err)
+
+	expected := &testpb.Reply{Content: "received message"}
+	assert.True(t, proto.Equal(expected, actual))
+
+	// create a message to send to the test actor
+	message = new(testpb.TestSend)
+	// send the message to the actor
+	for i := 0; i < 10; i++ {
+		err = RemoteTell(ctx, addr, message)
+		// perform some assertions
+		require.NoError(t, err)
+	}
+
+	count := 11
+	assert.EqualValues(t, count, actorRef.ReceivedCount(ctx))
+
+	// stop the actor after some time
+	ctx, cancel := context.WithTimeout(ctx, time.Second)
+	defer cancel()
+
+	err = sys.Stop(ctx)
+}
diff --git a/actors/helper_test.go b/actors/helper_test.go
index 4b3cc25..f846b63 100644
--- a/actors/helper_test.go
+++ b/actors/helper_test.go
@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/tochemey/goakt/log"
+	pb "github.com/tochemey/goakt/messages/v1"
 	testspb "github.com/tochemey/goakt/test/data/pb/v1"
 )
 
@@ -57,7 +58,7 @@ func (p *TestActor) PostStop(context.Context) error {
 
 // Receive processes any message dropped into the actor mailbox without a reply
 func (p *TestActor) Receive(ctx ReceiveContext) {
-	switch ctx.Message().(type) {
+	switch msg := ctx.Message().(type) {
 	case *testspb.TestSend:
 	case *testspb.TestPanic:
 		p.logger.Panic("Boom")
@@ -74,6 +75,20 @@ func (p *TestActor) Receive(ctx ReceiveContext) {
 		}()
 		// block until timer is up
 		wg.Wait()
+	case *pb.RemoteMessage:
+		p.logger.Info("received remote message")
+		// let us handle the actual message sent to the actor
+		remoteMsg, _ := msg.GetMessage().UnmarshalNew()
+		switch remoteMsg.(type) {
+		case *testspb.TestReply:
+			// define the message reply
+			reply := &testspb.Reply{Content: "received message"}
+			// send the message reply
+			ctx.Response(reply)
+		case *testspb.TestSend:
+			// pass
+		}
+
 	default:
 		p.logger.Panic(ErrUnhandled)
 	}
diff --git a/actors/option_test.go b/actors/option_test.go
index f952d24..0552343 100644
--- a/actors/option_test.go
+++ b/actors/option_test.go
@@ -50,6 +50,11 @@ func TestOptions(t *testing.T) {
 			option:   WithRemoting("localhost", 3100),
 			expected: actorSystem{remotingEnabled: atomic.NewBool(true), remotingPort: 3100, remotingHost: "localhost"},
 		},
+		{
+			name:     "WithShutdownTimeout",
+			option:   WithShutdownTimeout(2 * time.Second),
+			expected: actorSystem{shutdownTimeout: 2. * time.Second},
+		},
 	}
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
diff --git a/actors/pid_test.go b/actors/pid_test.go
index a2bddf7..59d8c76 100644
--- a/actors/pid_test.go
+++ b/actors/pid_test.go
@@ -24,6 +24,8 @@ func TestActorReceive(t *testing.T) {
 	defer goleak.VerifyNone(t,
 		goleak.IgnoreTopFunction("github.com/golang/glog.(*loggingT).flushDaemon"),
 		goleak.IgnoreTopFunction("github.com/go-redis/redis/v8/internal/pool.(*ConnPool).reaper"),
+		goleak.IgnoreTopFunction("golang.org/x/net/http2.(*serverConn).serve"),
+		goleak.IgnoreTopFunction("internal/poll.runtime_pollWait"),
 	)
 	ctx := context.TODO()
 
@@ -63,6 +65,8 @@ func TestActorWithPassivation(t *testing.T) {
 	defer goleak.VerifyNone(t,
 		goleak.IgnoreTopFunction("github.com/golang/glog.(*loggingT).flushDaemon"),
 		goleak.IgnoreTopFunction("github.com/go-redis/redis/v8/internal/pool.(*ConnPool).reaper"),
+		goleak.IgnoreTopFunction("golang.org/x/net/http2.(*serverConn).serve"),
+		goleak.IgnoreTopFunction("internal/poll.runtime_pollWait"),
 	)
 	ctx := context.TODO()
 	// create a Ping actor
@@ -96,6 +100,8 @@ func TestActorWithReply(t *testing.T) {
 		defer goleak.VerifyNone(t,
 			goleak.IgnoreTopFunction("github.com/golang/glog.(*loggingT).flushDaemon"),
 			goleak.IgnoreTopFunction("github.com/go-redis/redis/v8/internal/pool.(*ConnPool).reaper"),
+			goleak.IgnoreTopFunction("golang.org/x/net/http2.(*serverConn).serve"),
+			goleak.IgnoreTopFunction("internal/poll.runtime_pollWait"),
 		)
 		ctx := context.TODO()
 		// create a Ping actor
@@ -122,6 +128,8 @@ func TestActorWithReply(t *testing.T) {
 		defer goleak.VerifyNone(t,
 			goleak.IgnoreTopFunction("github.com/golang/glog.(*loggingT).flushDaemon"),
 			goleak.IgnoreTopFunction("github.com/go-redis/redis/v8/internal/pool.(*ConnPool).reaper"),
+			goleak.IgnoreTopFunction("golang.org/x/net/http2.(*serverConn).serve"),
+			goleak.IgnoreTopFunction("internal/poll.runtime_pollWait"),
 		)
 		ctx := context.TODO()
 		// create a Ping actor
@@ -149,6 +157,8 @@ func TestActorRestart(t *testing.T) {
 		defer goleak.VerifyNone(t,
 			goleak.IgnoreTopFunction("github.com/golang/glog.(*loggingT).flushDaemon"),
 			goleak.IgnoreTopFunction("github.com/go-redis/redis/v8/internal/pool.(*ConnPool).reaper"),
+			goleak.IgnoreTopFunction("golang.org/x/net/http2.(*serverConn).serve"),
+			goleak.IgnoreTopFunction("internal/poll.runtime_pollWait"),
 		)
 		ctx := context.TODO()
 
@@ -220,6 +230,8 @@ func TestActorRestart(t *testing.T) {
 		defer goleak.VerifyNone(t,
 			goleak.IgnoreTopFunction("github.com/golang/glog.(*loggingT).flushDaemon"),
 			goleak.IgnoreTopFunction("github.com/go-redis/redis/v8/internal/pool.(*ConnPool).reaper"),
+			goleak.IgnoreTopFunction("golang.org/x/net/http2.(*serverConn).serve"),
+			goleak.IgnoreTopFunction("internal/poll.runtime_pollWait"),
 		)
 		ctx := context.TODO()
 
@@ -265,6 +277,8 @@ func TestChildActor(t *testing.T) {
 		defer goleak.VerifyNone(t,
 			goleak.IgnoreTopFunction("github.com/golang/glog.(*loggingT).flushDaemon"),
 			goleak.IgnoreTopFunction("github.com/go-redis/redis/v8/internal/pool.(*ConnPool).reaper"),
+			goleak.IgnoreTopFunction("golang.org/x/net/http2.(*serverConn).serve"),
+			goleak.IgnoreTopFunction("internal/poll.runtime_pollWait"),
 		)
 
 		// create a test context
@@ -302,6 +316,8 @@ func TestChildActor(t *testing.T) {
 		defer goleak.VerifyNone(t,
 			goleak.IgnoreTopFunction("github.com/golang/glog.(*loggingT).flushDaemon"),
 			goleak.IgnoreTopFunction("github.com/go-redis/redis/v8/internal/pool.(*ConnPool).reaper"),
+			goleak.IgnoreTopFunction("golang.org/x/net/http2.(*serverConn).serve"),
+			goleak.IgnoreTopFunction("internal/poll.runtime_pollWait"),
 		)
 
 		// create a test context
@@ -343,6 +359,8 @@ func TestChildActor(t *testing.T) {
 		defer goleak.VerifyNone(t,
 			goleak.IgnoreTopFunction("github.com/golang/glog.(*loggingT).flushDaemon"),
 			goleak.IgnoreTopFunction("github.com/go-redis/redis/v8/internal/pool.(*ConnPool).reaper"),
+			goleak.IgnoreTopFunction("golang.org/x/net/http2.(*serverConn).serve"),
+			goleak.IgnoreTopFunction("internal/poll.runtime_pollWait"),
 		)
 
 		// create a test context