Repository: Tochemey/goakt
Pull request #366 (2024-06-15 21:33:47)

feat: PipeTo implementation

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/Tochemey/goakt/pull/366?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) Report
Attention: Patch coverage is `93.94572%` with `29 lines` in your changes missing coverage. Please review.
> Project coverage is 85.25%. Comparing base [(`fe716dc`)](https://app.codecov.io/gh/Tochemey/goakt/commit/fe716dce94c91295422318f4054fbd58d9251ac2?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) to head [(`48802ea`)](https://app.codecov.io/gh/Tochemey/goakt/commit/48802ea17fe482f13e6f06246e218b0e3adc15e7?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene).

| [Files](https://app.codecov.io/gh/Tochemey/goakt/pull/366?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) | Patch % | Lines |
|---|---|---|
| [actors/pid.go](https://app.codecov.io/gh/Tochemey/goakt/pull/366?src=pr&el=tree&filepath=actors%2Fpid.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene#diff-YWN0b3JzL3BpZC5nbw==) | 91.83% | [19 Missing and 10 partials :warning: ](https://app.codecov.io/gh/Tochemey/goakt/pull/366?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) |

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


```diff
@@            Coverage Diff             @@
##             main     #366      +/-   ##
==========================================
+ Coverage   85.01%   85.25%   +0.23%     
==========================================
  Files          36       37       +1     
  Lines        3237     3357     +120     
==========================================
+ Hits         2752     2862     +110     
- Misses        366      374       +8     
- Partials      119      121       +2     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/Tochemey/goakt/pull/366?dropdown=coverage&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 7a41032..1dfe5fa 100644
--- a/actors/actor_system.go
+++ b/actors/actor_system.go
@@ -429,7 +429,7 @@ func (x *actorSystem) Spawn(ctx context.Context, name string, actor Actor) (PID,
 	opts := []pidOption{
 		withInitMaxRetries(x.actorInitMaxRetries),
 		withPassivationAfter(x.expireActorAfter),
-		withSendReplyTimeout(x.replyTimeout),
+		withReplyTimeout(x.replyTimeout),
 		withCustomLogger(x.logger),
 		withActorSystem(x),
 		withSupervisorStrategy(x.supervisorStrategy),
@@ -503,7 +503,7 @@ func (x *actorSystem) SpawnFromFunc(ctx context.Context, receiveFunc ReceiveFunc
 	pidOpts := []pidOption{
 		withInitMaxRetries(x.actorInitMaxRetries),
 		withPassivationAfter(x.expireActorAfter),
-		withSendReplyTimeout(x.replyTimeout),
+		withReplyTimeout(x.replyTimeout),
 		withCustomLogger(x.logger),
 		withActorSystem(x),
 		withSupervisorStrategy(x.supervisorStrategy),
diff --git a/actors/context.go b/actors/context.go
index 8c71eb0..c88d079 100644
--- a/actors/context.go
+++ b/actors/context.go
@@ -33,6 +33,7 @@ import (
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/anypb"
 
+	"github.com/tochemey/goakt/v2/future"
 	"github.com/tochemey/goakt/v2/goaktpb"
 )
 
@@ -134,6 +135,11 @@ type ReceiveContext interface {
 	// Err is used instead of panicking within a message handler.
 	// One can also call panic which is not the recommended way
 	Err(err error)
+	// PipeTo processes a long-running task and pipes the result to the provided actor.
+	// The successful result of the task will be put onto the provided actor mailbox.
+	// This is useful when interacting with external services.
+	// It’s common that you would like to use the value of the response in the actor when the long-running task is completed
+	PipeTo(to PID, task future.Task)
 }
 
 type receiveContext struct {
@@ -179,8 +185,9 @@ func (c *receiveContext) WithRemoteSender(remoteSender *goaktpb.Address) *receiv
 // Self returns the receiver PID of the message
 func (c *receiveContext) Self() PID {
 	c.mu.Lock()
-	defer c.mu.Unlock()
-	return c.recipient
+	self := c.recipient
+	c.mu.Unlock()
+	return self
 }
 
 // Err is used instead of panicking within a message handler.
@@ -194,9 +201,9 @@ func (c *receiveContext) Err(err error) {
 func (c *receiveContext) Response(resp proto.Message) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	defer close(c.response)
 	// only set a response when the message is sync message
 	if !c.isAsyncMessage {
+		defer close(c.response)
 		c.response <- resp
 	}
 }
@@ -204,30 +211,34 @@ func (c *receiveContext) Response(resp proto.Message) {
 // Context represents the context attached to the message
 func (c *receiveContext) Context() context.Context {
 	c.mu.Lock()
-	defer c.mu.Unlock()
-	return c.ctx
+	ctx := c.ctx
+	c.mu.Unlock()
+	return ctx
 }
 
 // Sender of the message
 func (c *receiveContext) Sender() PID {
 	c.mu.Lock()
-	defer c.mu.Unlock()
-	return c.sender
+	sender := c.sender
+	c.mu.Unlock()
+	return sender
 }
 
 // RemoteSender defines the remote sender of the message if it is a remote message
 // This is set to RemoteNoSender when the message is not a remote message
 func (c *receiveContext) RemoteSender() *goaktpb.Address {
 	c.mu.Lock()
-	defer c.mu.Unlock()
-	return c.remoteSender
+	sender := c.remoteSender
+	c.mu.Unlock()
+	return sender
 }
 
 // Message is the actual message sent
 func (c *receiveContext) Message() proto.Message {
 	c.mu.Lock()
-	defer c.mu.Unlock()
-	return c.message
+	message := c.message
+	c.mu.Unlock()
+	return message
 }
 
 // BecomeStacked sets a new behavior to the actor.
@@ -266,7 +277,7 @@ func (c *receiveContext) Become(behavior Behavior) {
 func (c *receiveContext) Stash() {
 	c.mu.Lock()
 	recipient := c.recipient
-	defer c.mu.Unlock()
+	c.mu.Unlock()
 	if err := recipient.stash(c); err != nil {
 		c.Err(err)
 	}
@@ -276,7 +287,7 @@ func (c *receiveContext) Stash() {
 func (c *receiveContext) Unstash() {
 	c.mu.Lock()
 	recipient := c.recipient
-	defer c.mu.Unlock()
+	c.mu.Unlock()
 	if err := recipient.unstash(); err != nil {
 		c.Err(err)
 	}
@@ -287,7 +298,7 @@ func (c *receiveContext) Unstash() {
 func (c *receiveContext) UnstashAll() {
 	c.mu.Lock()
 	recipient := c.recipient
-	defer c.mu.Unlock()
+	c.mu.Unlock()
 	if err := recipient.unstashAll(); err != nil {
 		c.Err(err)
 	}
@@ -297,8 +308,8 @@ func (c *receiveContext) UnstashAll() {
 func (c *receiveContext) Tell(to PID, message proto.Message) {
 	c.mu.Lock()
 	recipient := c.recipient
-	defer c.mu.Unlock()
 	ctx := context.WithoutCancel(c.ctx)
+	c.mu.Unlock()
 	if err := recipient.Tell(ctx, to, message); err != nil {
 		panic(err)
 	}
@@ -311,8 +322,8 @@ func (c *receiveContext) Tell(to PID, message proto.Message) {
 func (c *receiveContext) BatchTell(to PID, messages ...proto.Message) {
 	c.mu.Lock()
 	recipient := c.recipient
-	defer c.mu.Unlock()
 	ctx := context.WithoutCancel(c.ctx)
+	c.mu.Unlock()
 	if err := recipient.BatchTell(ctx, to, messages...); err != nil {
 		c.Err(err)
 	}
@@ -324,8 +335,8 @@ func (c *receiveContext) BatchTell(to PID, messages ...proto.Message) {
 func (c *receiveContext) Ask(to PID, message proto.Message) (response proto.Message) {
 	c.mu.Lock()
 	recipient := c.recipient
-	defer c.mu.Unlock()
 	ctx := context.WithoutCancel(c.ctx)
+	c.mu.Unlock()
 	reply, err := recipient.Ask(ctx, to, message)
 	if err != nil {
 		c.Err(err)
@@ -339,8 +350,8 @@ func (c *receiveContext) Ask(to PID, message proto.Message) (response proto.Mess
 func (c *receiveContext) BatchAsk(to PID, messages ...proto.Message) (responses chan proto.Message) {
 	c.mu.Lock()
 	recipient := c.recipient
-	defer c.mu.Unlock()
 	ctx := context.WithoutCancel(c.ctx)
+	c.mu.Unlock()
 	reply, err := recipient.BatchAsk(ctx, to, messages...)
 	if err != nil {
 		c.Err(err)
@@ -352,8 +363,8 @@ func (c *receiveContext) BatchAsk(to PID, messages ...proto.Message) (responses
 func (c *receiveContext) RemoteTell(to *goaktpb.Address, message proto.Message) {
 	c.mu.Lock()
 	recipient := c.recipient
-	defer c.mu.Unlock()
 	ctx := context.WithoutCancel(c.ctx)
+	c.mu.Unlock()
 	if err := recipient.RemoteTell(ctx, to, message); err != nil {
 		c.Err(err)
 	}
@@ -364,8 +375,8 @@ func (c *receiveContext) RemoteTell(to *goaktpb.Address, message proto.Message)
 func (c *receiveContext) RemoteAsk(to *goaktpb.Address, message proto.Message) (response *anypb.Any) {
 	c.mu.Lock()
 	recipient := c.recipient
-	defer c.mu.Unlock()
 	ctx := context.WithoutCancel(c.ctx)
+	c.mu.Unlock()
 	reply, err := recipient.RemoteAsk(ctx, to, message)
 	if err != nil {
 		c.Err(err)
@@ -378,8 +389,8 @@ func (c *receiveContext) RemoteAsk(to *goaktpb.Address, message proto.Message) (
 func (c *receiveContext) RemoteBatchTell(to *goaktpb.Address, messages ...proto.Message) {
 	c.mu.Lock()
 	recipient := c.recipient
-	defer c.mu.Unlock()
 	ctx := context.WithoutCancel(c.ctx)
+	c.mu.Unlock()
 	if err := recipient.RemoteBatchTell(ctx, to, messages...); err != nil {
 		c.Err(err)
 	}
@@ -391,8 +402,8 @@ func (c *receiveContext) RemoteBatchTell(to *goaktpb.Address, messages ...proto.
 func (c *receiveContext) RemoteBatchAsk(to *goaktpb.Address, messages ...proto.Message) (responses []*anypb.Any) {
 	c.mu.Lock()
 	recipient := c.recipient
-	defer c.mu.Unlock()
 	ctx := context.WithoutCancel(c.ctx)
+	c.mu.Unlock()
 	replies, err := recipient.RemoteBatchAsk(ctx, to, messages...)
 	if err != nil {
 		c.Err(err)
@@ -405,8 +416,8 @@ func (c *receiveContext) RemoteBatchAsk(to *goaktpb.Address, messages ...proto.M
 func (c *receiveContext) RemoteLookup(host string, port int, name string) (addr *goaktpb.Address) {
 	c.mu.Lock()
 	recipient := c.recipient
-	defer c.mu.Unlock()
 	ctx := context.WithoutCancel(c.ctx)
+	c.mu.Unlock()
 	remoteAddr, err := recipient.RemoteLookup(ctx, host, port, name)
 	if err != nil {
 		c.Err(err)
@@ -420,8 +431,8 @@ func (c *receiveContext) RemoteLookup(host string, port int, name string) (addr
 func (c *receiveContext) Shutdown() {
 	c.mu.Lock()
 	recipient := c.recipient
-	defer c.mu.Unlock()
 	ctx := context.WithoutCancel(c.ctx)
+	c.mu.Unlock()
 	if err := recipient.Shutdown(ctx); err != nil {
 		c.Err(err)
 	}
@@ -431,8 +442,8 @@ func (c *receiveContext) Shutdown() {
 func (c *receiveContext) Spawn(name string, actor Actor) PID {
 	c.mu.Lock()
 	recipient := c.recipient
-	defer c.mu.Unlock()
 	ctx := context.WithoutCancel(c.ctx)
+	c.mu.Unlock()
 	pid, err := recipient.SpawnChild(ctx, name, actor)
 	if err != nil {
 		c.Err(err)
@@ -451,7 +462,7 @@ func (c *receiveContext) Children() []PID {
 func (c *receiveContext) Child(name string) PID {
 	c.mu.Lock()
 	recipient := c.recipient
-	defer c.mu.Unlock()
+	c.mu.Unlock()
 	pid, err := recipient.Child(name)
 	if err != nil {
 		c.Err(err)
@@ -464,11 +475,9 @@ func (c *receiveContext) Child(name string) PID {
 func (c *receiveContext) Stop(child PID) {
 	c.mu.Lock()
 	recipient := c.recipient
-	defer c.mu.Unlock()
-
 	ctx := context.WithoutCancel(c.ctx)
-	err := recipient.Stop(ctx, child)
-	if err != nil {
+	c.mu.Unlock()
+	if err := recipient.Stop(ctx, child); err != nil {
 		c.Err(err)
 	}
 }
@@ -509,10 +518,23 @@ func (c *receiveContext) Unhandled() {
 func (c *receiveContext) RemoteReSpawn(host string, port int, name string) {
 	c.mu.Lock()
 	recipient := c.recipient
-	defer c.mu.Unlock()
 	ctx := context.WithoutCancel(c.ctx)
-	err := recipient.RemoteReSpawn(ctx, host, port, name)
-	if err != nil {
+	c.mu.Unlock()
+	if err := recipient.RemoteReSpawn(ctx, host, port, name); err != nil {
+		c.Err(err)
+	}
+}
+
+// PipeTo processes a long-running task and pipes the result to the provided actor.
+// The successful result of the task will be put onto the provided actor mailbox.
+// This is useful when interacting with external services.
+// It’s common that you would like to use the value of the response in the actor when the long-running task is completed
+func (c *receiveContext) PipeTo(to PID, task future.Task) {
+	c.mu.Lock()
+	recipient := c.recipient
+	ctx := context.WithoutCancel(c.ctx)
+	c.mu.Unlock()
+	if err := recipient.PipeTo(ctx, to, task); err != nil {
 		c.Err(err)
 	}
 }
diff --git a/actors/errors.go b/actors/errors.go
index 75872ab..29c93af 100644
--- a/actors/errors.go
+++ b/actors/errors.go
@@ -83,6 +83,8 @@ var (
 	ErrPeerNotFound = errors.New("peer is not found")
 	// ErrKindsRequired is returned when the actor kinds are not provided when enabling the cluster mode
 	ErrKindsRequired = errors.New("actor kinds are required")
+	// ErrUndefinedTask is returned when piping a long-running task result to an actor
+	ErrUndefinedTask = errors.New("task is not defined")
 )
 
 // IsEOF returns true if the given error is an EOF error
diff --git a/actors/pid.go b/actors/pid.go
index 5dc520e..90a6a0a 100644
--- a/actors/pid.go
+++ b/actors/pid.go
@@ -45,6 +45,7 @@ import (
 	"google.golang.org/protobuf/types/known/anypb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
+	"github.com/tochemey/goakt/v2/future"
 	"github.com/tochemey/goakt/v2/goaktpb"
 	"github.com/tochemey/goakt/v2/internal/eventstream"
 	"github.com/tochemey/goakt/v2/internal/http"
@@ -65,6 +66,13 @@ type watcher struct {
 	Done      chan types.Unit // Done when watching is completed
 }
 
+// completion is used to track completions' completion
+// to pipe the result to the appropriate PID
+type completion struct {
+	Receiver PID
+	Task     future.Task
+}
+
 // PID defines the various actions one can perform on a given actor
 type PID interface {
 	// ID is a convenient method that returns the actor unique identifier
@@ -141,6 +149,11 @@ type PID interface {
 	StashSize() uint64
 	// Equals is a convenient method to compare two PIDs
 	Equals(to PID) bool
+	// PipeTo processes a long-running task and pipes the result to the provided actor.
+	// The successful result of the task will be put onto the provided actor mailbox.
+	// This is useful when interacting with external services.
+	// It’s common that you would like to use the value of the response in the actor when the long-running task is completed
+	PipeTo(ctx context.Context, to PID, task future.Task) error
 	// push a message to the actor's receiveContextBuffer
 	doReceive(ctx ReceiveContext)
 	// watchers returns the list of watchMen
@@ -349,43 +362,43 @@ func newPID(ctx context.Context, actorPath *Path, actor Actor, opts ...pidOption
 
 // ID is a convenient method that returns the actor unique identifier
 // An actor unique identifier is its address in the actor system.
-func (p *pid) ID() string {
-	return p.ActorPath().String()
+func (x *pid) ID() string {
+	return x.ActorPath().String()
 }
 
 // Name returns the actor given name
-func (p *pid) Name() string {
-	return p.ActorPath().Name()
+func (x *pid) Name() string {
+	return x.ActorPath().Name()
 }
 
 // Equals is a convenient method to compare two PIDs
-func (p *pid) Equals(to PID) bool {
-	return strings.EqualFold(p.ID(), to.ID())
+func (x *pid) Equals(to PID) bool {
+	return strings.EqualFold(x.ID(), to.ID())
 }
 
 // ActorHandle returns the underlying Actor
-func (p *pid) ActorHandle() Actor {
-	return p.Actor
+func (x *pid) ActorHandle() Actor {
+	return x.Actor
 }
 
 // Child returns the named child actor if it is alive
-func (p *pid) Child(name string) (PID, error) {
-	if !p.IsRunning() {
+func (x *pid) Child(name string) (PID, error) {
+	if !x.IsRunning() {
 		return nil, ErrDead
 	}
-	childActorPath := NewPath(name, p.ActorPath().Address()).WithParent(p.ActorPath())
-	if cid, ok := p.children.get(childActorPath); ok {
-		p.childrenCount.Inc()
+	childActorPath := NewPath(name, x.ActorPath().Address()).WithParent(x.ActorPath())
+	if cid, ok := x.children.get(childActorPath); ok {
+		x.childrenCount.Inc()
 		return cid, nil
 	}
 	return nil, ErrActorNotFound(childActorPath.String())
 }
 
 // Children returns the list of all the children of the given actor that are still alive or an empty list
-func (p *pid) Children() []PID {
-	p.rwLocker.RLock()
-	children := p.children.pids()
-	p.rwLocker.RUnlock()
+func (x *pid) Children() []PID {
+	x.rwLocker.RLock()
+	children := x.children.pids()
+	x.rwLocker.RUnlock()
 
 	cids := make([]PID, 0, len(children))
 	for _, child := range children {
@@ -399,11 +412,11 @@ func (p *pid) Children() []PID {
 
 // Stop forces the child Actor under the given name to terminate after it finishes processing its current message.
 // Nothing happens if child is already stopped.
-func (p *pid) Stop(ctx context.Context, cid PID) error {
-	spanCtx, span := p.tracer.Start(ctx, "Stop")
+func (x *pid) Stop(ctx context.Context, cid PID) error {
+	spanCtx, span := x.tracer.Start(ctx, "Stop")
 	defer span.End()
 
-	if !p.IsRunning() {
+	if !x.IsRunning() {
 		span.SetStatus(codes.Error, "Stop")
 		span.RecordError(ErrDead)
 		return ErrDead
@@ -415,9 +428,9 @@ func (p *pid) Stop(ctx context.Context, cid PID) error {
 		return ErrUndefinedActor
 	}
 
-	p.rwLocker.RLock()
-	children := p.children
-	p.rwLocker.RUnlock()
+	x.rwLocker.RLock()
+	children := x.children
+	x.rwLocker.RUnlock()
 
 	if cid, ok := children.get(cid.ActorPath()); ok {
 		desc := fmt.Sprintf("child.[%s] Shutdown", cid.ActorPath())
@@ -439,49 +452,49 @@ func (p *pid) Stop(ctx context.Context, cid PID) error {
 
 // IsRunning returns true when the actor is alive ready to process messages and false
 // when the actor is stopped or not started at all
-func (p *pid) IsRunning() bool {
-	return p != nil && p != NoSender && p.isRunning.Load()
+func (x *pid) IsRunning() bool {
+	return x != nil && x != NoSender && x.isRunning.Load()
 }
 
 // ActorSystem returns the actor system
-func (p *pid) ActorSystem() ActorSystem {
-	p.rwLocker.RLock()
-	sys := p.system
-	p.rwLocker.RUnlock()
+func (x *pid) ActorSystem() ActorSystem {
+	x.rwLocker.RLock()
+	sys := x.system
+	x.rwLocker.RUnlock()
 	return sys
 }
 
 // ActorPath returns the path of the actor
-func (p *pid) ActorPath() *Path {
-	p.rwLocker.RLock()
-	path := p.actorPath
-	p.rwLocker.RUnlock()
+func (x *pid) ActorPath() *Path {
+	x.rwLocker.RLock()
+	path := x.actorPath
+	x.rwLocker.RUnlock()
 	return path
 }
 
 // Restart restarts the actor.
 // During restart all messages that are in the mailbox and not yet processed will be ignored
-func (p *pid) Restart(ctx context.Context) error {
-	spanCtx, span := p.tracer.Start(ctx, "Restart")
+func (x *pid) Restart(ctx context.Context) error {
+	spanCtx, span := x.tracer.Start(ctx, "Restart")
 	defer span.End()
 
-	if p == nil || p.ActorPath() == nil {
+	if x == nil || x.ActorPath() == nil {
 		span.SetStatus(codes.Error, "Restart")
 		span.RecordError(ErrUndefinedActor)
 		return ErrUndefinedActor
 	}
 
-	p.logger.Debugf("restarting actor=(%s)", p.actorPath.String())
+	x.logger.Debugf("restarting actor=(%s)", x.actorPath.String())
 
-	if p.IsRunning() {
-		if err := p.Shutdown(spanCtx); err != nil {
+	if x.IsRunning() {
+		if err := x.Shutdown(spanCtx); err != nil {
 			return err
 		}
 		ticker := time.NewTicker(10 * time.Millisecond)
 		tickerStopSig := make(chan types.Unit, 1)
 		go func() {
 			for range ticker.C {
-				if !p.IsRunning() {
+				if !x.IsRunning() {
 					tickerStopSig <- types.Unit{}
 					return
 				}
@@ -491,22 +504,22 @@ func (p *pid) Restart(ctx context.Context) error {
 		ticker.Stop()
 	}
 
-	p.mailbox.Reset()
-	p.resetBehavior()
-	if err := p.init(spanCtx); err != nil {
+	x.mailbox.Reset()
+	x.resetBehavior()
+	if err := x.init(spanCtx); err != nil {
 		return err
 	}
-	go p.receive()
-	if p.passivateAfter.Load() > 0 {
-		go p.passivationListener()
+	go x.receive()
+	if x.passivateAfter.Load() > 0 {
+		go x.passivationListener()
 	}
 
 	span.SetStatus(codes.Ok, "Restart")
-	p.restartCount.Inc()
+	x.restartCount.Inc()
 
-	if p.eventsStream != nil {
-		p.eventsStream.Publish(eventsTopic, &goaktpb.ActorRestarted{
-			Address:     p.ActorPath().RemoteAddress(),
+	if x.eventsStream != nil {
+		x.eventsStream.Publish(eventsTopic, &goaktpb.ActorRestarted{
+			Address:     x.ActorPath().RemoteAddress(),
 			RestartedAt: timestamppb.Now(),
 		})
 	}
@@ -516,50 +529,50 @@ func (p *pid) Restart(ctx context.Context) error {
 
 // SpawnChild creates a child actor and start watching it for error
 // When the given child actor already exists its PID will only be returned
-func (p *pid) SpawnChild(ctx context.Context, name string, actor Actor) (PID, error) {
-	spanCtx, span := p.tracer.Start(ctx, "SpawnChild")
+func (x *pid) SpawnChild(ctx context.Context, name string, actor Actor) (PID, error) {
+	spanCtx, span := x.tracer.Start(ctx, "SpawnChild")
 	defer span.End()
 
-	if !p.IsRunning() {
+	if !x.IsRunning() {
 		span.SetStatus(codes.Error, "SpawnChild")
 		span.RecordError(ErrDead)
 		return nil, ErrDead
 	}
 
-	childActorPath := NewPath(name, p.ActorPath().Address()).WithParent(p.ActorPath())
+	childActorPath := NewPath(name, x.ActorPath().Address()).WithParent(x.ActorPath())
 
-	p.rwLocker.RLock()
-	children := p.children
-	p.rwLocker.RUnlock()
+	x.rwLocker.RLock()
+	children := x.children
+	x.rwLocker.RUnlock()
 
 	if cid, ok := children.get(childActorPath); ok {
 		return cid, nil
 	}
 
-	p.rwLocker.Lock()
-	defer p.rwLocker.Unlock()
+	x.rwLocker.Lock()
+	defer x.rwLocker.Unlock()
 
 	// create the child actor options child inherit parent's options
 	opts := []pidOption{
-		withInitMaxRetries(int(p.initMaxRetries.Load())),
-		withPassivationAfter(p.passivateAfter.Load()),
-		withSendReplyTimeout(p.replyTimeout.Load()),
-		withCustomLogger(p.logger),
-		withActorSystem(p.system),
-		withSupervisorStrategy(p.supervisorStrategy),
-		withMailboxSize(p.mailboxSize),
-		withStash(p.stashCapacity.Load()),
-		withMailbox(p.mailbox.Clone()),
-		withEventsStream(p.eventsStream),
-		withInitTimeout(p.initTimeout.Load()),
-		withShutdownTimeout(p.shutdownTimeout.Load()),
-	}
-
-	if p.traceEnabled.Load() {
+		withInitMaxRetries(int(x.initMaxRetries.Load())),
+		withPassivationAfter(x.passivateAfter.Load()),
+		withReplyTimeout(x.replyTimeout.Load()),
+		withCustomLogger(x.logger),
+		withActorSystem(x.system),
+		withSupervisorStrategy(x.supervisorStrategy),
+		withMailboxSize(x.mailboxSize),
+		withStash(x.stashCapacity.Load()),
+		withMailbox(x.mailbox.Clone()),
+		withEventsStream(x.eventsStream),
+		withInitTimeout(x.initTimeout.Load()),
+		withShutdownTimeout(x.shutdownTimeout.Load()),
+	}
+
+	if x.traceEnabled.Load() {
 		opts = append(opts, withTracing())
 	}
 
-	if p.metricEnabled.Load() {
+	if x.metricEnabled.Load() {
 		opts = append(opts, withMetric())
 	}
 
@@ -575,16 +588,16 @@ func (p *pid) SpawnChild(ctx context.Context, name string, actor Actor) (PID, er
 		return nil, err
 	}
 
-	p.children.set(cid)
-	p.Watch(cid)
+	x.children.set(cid)
+	x.Watch(cid)
 
 	span.SetStatus(codes.Ok, "SpawnChild")
 
-	if p.eventsStream != nil {
-		p.eventsStream.Publish(eventsTopic, &goaktpb.ActorChildCreated{
+	if x.eventsStream != nil {
+		x.eventsStream.Publish(eventsTopic, &goaktpb.ActorChildCreated{
 			Address:   cid.ActorPath().RemoteAddress(),
 			CreatedAt: timestamppb.Now(),
-			Parent:    p.ActorPath().RemoteAddress(),
+			Parent:    x.ActorPath().RemoteAddress(),
 		})
 	}
 
@@ -592,17 +605,45 @@ func (p *pid) SpawnChild(ctx context.Context, name string, actor Actor) (PID, er
 }
 
 // StashSize returns the stash buffer size
-func (p *pid) StashSize() uint64 {
-	if p.stashBuffer == nil {
+func (x *pid) StashSize() uint64 {
+	if x.stashBuffer == nil {
 		return 0
 	}
-	return p.stashBuffer.Size()
+	return x.stashBuffer.Size()
+}
+
+// PipeTo processes a long-running task and pipes the result to the provided actor.
+// The successful result of the task will be put onto the provided actor mailbox.
+// This is useful when interacting with external services.
+// It’s common that you would like to use the value of the response in the actor when the long-running task is completed
+func (x *pid) PipeTo(ctx context.Context, to PID, task future.Task) error {
+	ctx, span := x.tracer.Start(ctx, "PipeTo")
+	defer span.End()
+
+	if task == nil {
+		span.SetStatus(codes.Error, "PipeTo")
+		span.RecordError(ErrUndefinedTask)
+		return ErrUndefinedTask
+	}
+
+	if !to.IsRunning() {
+		span.SetStatus(codes.Error, "PipeTo")
+		span.RecordError(ErrDead)
+		return ErrDead
+	}
+
+	go x.handleCompletion(ctx, &completion{
+		Receiver: to,
+		Task:     task,
+	})
+
+	return nil
 }
 
 // Ask sends a synchronous message to another actor and expect a response.
 // This block until a response is received or timed out.
-func (p *pid) Ask(ctx context.Context, to PID, message proto.Message) (response proto.Message, err error) {
-	spanCtx, span := p.tracer.Start(ctx, "Ask")
+func (x *pid) Ask(ctx context.Context, to PID, message proto.Message) (response proto.Message, err error) {
+	spanCtx, span := x.tracer.Start(ctx, "Ask")
 	defer span.End()
 
 	if !to.IsRunning() {
@@ -611,29 +652,29 @@ func (p *pid) Ask(ctx context.Context, to PID, message proto.Message) (response
 		return nil, ErrDead
 	}
 
-	messageContext := newReceiveContext(spanCtx, p, to, message, false)
+	messageContext := newReceiveContext(spanCtx, x, to, message, false)
 	to.doReceive(messageContext)
 
-	for await := time.After(p.replyTimeout.Load()); ; {
+	for await := time.After(x.replyTimeout.Load()); ; {
 		select {
 		case response = <-messageContext.response:
-			p.lastProcessingDuration.Store(time.Since(p.lastProcessingTime.Load()))
+			x.lastProcessingDuration.Store(time.Since(x.lastProcessingTime.Load()))
 			span.SetStatus(codes.Ok, "Ask")
 			return
 		case <-await:
-			p.lastProcessingDuration.Store(time.Since(p.lastProcessingTime.Load()))
+			x.lastProcessingDuration.Store(time.Since(x.lastProcessingTime.Load()))
 			err = ErrRequestTimeout
 			span.SetStatus(codes.Error, "Ask")
 			span.RecordError(err)
-			p.handleError(messageContext, err)
+			x.handleError(messageContext, err)
 			return nil, err
 		}
 	}
 }
 
 // Tell sends an asynchronous message to another PID
-func (p *pid) Tell(ctx context.Context, to PID, message proto.Message) error {
-	spanCtx, span := p.tracer.Start(ctx, "Tell")
+func (x *pid) Tell(ctx context.Context, to PID, message proto.Message) error {
+	spanCtx, span := x.tracer.Start(ctx, "Tell")
 	defer span.End()
 
 	if !to.IsRunning() {
@@ -642,9 +683,9 @@ func (p *pid) Tell(ctx context.Context, to PID, message proto.Message) error {
 		return ErrDead
 	}
 
-	messageContext := newReceiveContext(spanCtx, p, to, message, true)
+	messageContext := newReceiveContext(spanCtx, x, to, message, true)
 	to.doReceive(messageContext)
-	p.lastProcessingDuration.Store(time.Since(p.lastProcessingTime.Load()))
+	x.lastProcessingDuration.Store(time.Since(x.lastProcessingTime.Load()))
 	span.SetStatus(codes.Ok, "Tell")
 	return nil
 }
@@ -653,8 +694,8 @@ func (p *pid) Tell(ctx context.Context, to PID, message proto.Message) error {
 // The messages will be processed one after the other in the order they are sent.
 // This is a design choice to follow the simple principle of one message at a time processing by actors.
 // When BatchTell encounter a single message it will fall back to a Tell call.
-func (p *pid) BatchTell(ctx context.Context, to PID, messages ...proto.Message) error {
-	spanCtx, span := p.tracer.Start(ctx, "BatchTell")
+func (x *pid) BatchTell(ctx context.Context, to PID, messages ...proto.Message) error {
+	spanCtx, span := x.tracer.Start(ctx, "BatchTell")
 	defer span.End()
 
 	if !to.IsRunning() {
@@ -665,16 +706,16 @@ func (p *pid) BatchTell(ctx context.Context, to PID, messages ...proto.Message)
 
 	if len(messages) == 1 {
 		// no need to record span error here because Tell handles it
-		return p.Tell(spanCtx, to, messages[0])
+		return x.Tell(spanCtx, to, messages[0])
 	}
 
 	for i := 0; i < len(messages); i++ {
 		message := messages[i]
-		messageContext := newReceiveContext(ctx, p, to, message, true)
+		messageContext := newReceiveContext(ctx, x, to, message, true)
 		to.doReceive(messageContext)
 	}
 
-	p.lastProcessingDuration.Store(time.Since(p.lastProcessingTime.Load()))
+	x.lastProcessingDuration.Store(time.Since(x.lastProcessingTime.Load()))
 	span.SetStatus(codes.Ok, "BatchTell")
 	return nil
 }
@@ -682,8 +723,8 @@ func (p *pid) BatchTell(ctx context.Context, to PID, messages ...proto.Message)
 // BatchAsk sends a synchronous bunch of messages to the given PID and expect responses in the same order as the messages.
 // The messages will be processed one after the other in the order they are sent.
 // This is a design choice to follow the simple principle of one message at a time processing by actors.
-func (p *pid) BatchAsk(ctx context.Context, to PID, messages ...proto.Message) (responses chan proto.Message, err error) {
-	spanCtx, span := p.tracer.Start(ctx, "BatchAsk")
+func (x *pid) BatchAsk(ctx context.Context, to PID, messages ...proto.Message) (responses chan proto.Message, err error) {
+	spanCtx, span := x.tracer.Start(ctx, "BatchAsk")
 	defer span.End()
 
 	if !to.IsRunning() {
@@ -697,24 +738,24 @@ func (p *pid) BatchAsk(ctx context.Context, to PID, messages ...proto.Message) (
 
 	for i := 0; i < len(messages); i++ {
 		message := messages[i]
-		messageContext := newReceiveContext(spanCtx, p, to, message, false)
+		messageContext := newReceiveContext(spanCtx, x, to, message, false)
 		to.doReceive(messageContext)
 
 		// await patiently to receive the response from the actor
 	timerLoop:
-		for await := time.After(p.replyTimeout.Load()); ; {
+		for await := time.After(x.replyTimeout.Load()); ; {
 			select {
 			case resp := <-messageContext.response:
-				p.lastProcessingDuration.Store(time.Since(p.lastProcessingTime.Load()))
+				x.lastProcessingDuration.Store(time.Since(x.lastProcessingTime.Load()))
 				responses <- resp
 				span.SetStatus(codes.Ok, "BatchAsk")
 				break timerLoop
 			case <-await:
-				p.lastProcessingDuration.Store(time.Since(p.lastProcessingTime.Load()))
+				x.lastProcessingDuration.Store(time.Since(x.lastProcessingTime.Load()))
 				err = ErrRequestTimeout
 				span.SetStatus(codes.Error, "BatchAsk")
 				span.RecordError(err)
-				p.handleError(messageContext, err)
+				x.handleError(messageContext, err)
 				return nil, err
 			}
 		}
@@ -723,8 +764,8 @@ func (p *pid) BatchAsk(ctx context.Context, to PID, messages ...proto.Message) (
 }
 
 // RemoteLookup look for an actor address on a remote node.
-func (p *pid) RemoteLookup(ctx context.Context, host string, port int, name string) (addr *goaktpb.Address, err error) {
-	remoteClient, err := p.getRemoteServiceClient(host, port)
+func (x *pid) RemoteLookup(ctx context.Context, host string, port int, name string) (addr *goaktpb.Address, err error) {
+	remoteClient, err := x.getRemoteServiceClient(host, port)
 	if err != nil {
 		return nil, err
 	}
@@ -748,22 +789,22 @@ func (p *pid) RemoteLookup(ctx context.Context, host string, port int, name stri
 }
 
 // RemoteTell sends a message to an actor remotely without expecting any reply
-func (p *pid) RemoteTell(ctx context.Context, to *goaktpb.Address, message proto.Message) error {
+func (x *pid) RemoteTell(ctx context.Context, to *goaktpb.Address, message proto.Message) error {
 	marshaled, err := anypb.New(message)
 	if err != nil {
 		return err
 	}
 
-	remoteService, err := p.getRemoteServiceClient(to.GetHost(), int(to.GetPort()))
+	remoteService, err := x.getRemoteServiceClient(to.GetHost(), int(to.GetPort()))
 	if err != nil {
 		return err
 	}
 
 	sender := &goaktpb.Address{
-		Host: p.ActorPath().Address().Host(),
-		Port: int32(p.ActorPath().Address().Port()),
-		Name: p.ActorPath().Name(),
-		Id:   p.ActorPath().ID().String(),
+		Host: x.ActorPath().Address().Host(),
+		Port: int32(x.ActorPath().Address().Port()),
+		Name: x.ActorPath().Name(),
+		Id:   x.ActorPath().ID().String(),
 	}
 
 	request := &internalpb.RemoteTellRequest{
@@ -774,7 +815,7 @@ func (p *pid) RemoteTell(ctx context.Context, to *goaktpb.Address, message proto
 		},
 	}
 
-	p.logger.Debugf("sending a message to remote=(%s:%d)", to.GetHost(), to.GetPort())
+	x.logger.Debugf("sending a message to remote=(%s:%d)", to.GetHost(), to.GetPort())
 	stream := remoteService.RemoteTell(ctx)
 	if err := stream.Send(request); err != nil {
 		if IsEOF(err) {
@@ -783,32 +824,32 @@ func (p *pid) RemoteTell(ctx context.Context, to *goaktpb.Address, message proto
 			}
 			return nil
 		}
-		p.logger.Error(errors.Wrapf(err, "failed to send message to remote=(%s:%d)", to.GetHost(), to.GetPort()))
+		x.logger.Error(errors.Wrapf(err, "failed to send message to remote=(%s:%d)", to.GetHost(), to.GetPort()))
 		return err
 	}
 
 	if _, err := stream.CloseAndReceive(); err != nil {
-		p.logger.Error(errors.Wrapf(err, "failed to send message to remote=(%s:%d)", to.GetHost(), to.GetPort()))
+		x.logger.Error(errors.Wrapf(err, "failed to send message to remote=(%s:%d)", to.GetHost(), to.GetPort()))
 		return err
 	}
 
-	p.logger.Debugf("message successfully sent to remote=(%s:%d)", to.GetHost(), to.GetPort())
+	x.logger.Debugf("message successfully sent to remote=(%s:%d)", to.GetHost(), to.GetPort())
 	return nil
 }
 
 // RemoteAsk sends a synchronous message to another actor remotely and expect a response.
-func (p *pid) RemoteAsk(ctx context.Context, to *goaktpb.Address, message proto.Message) (response *anypb.Any, err error) {
+func (x *pid) RemoteAsk(ctx context.Context, to *goaktpb.Address, message proto.Message) (response *anypb.Any, err error) {
 	marshaled, err := anypb.New(message)
 	if err != nil {
 		return nil, err
 	}
 
-	remoteService, err := p.getRemoteServiceClient(to.GetHost(), int(to.GetPort()))
+	remoteService, err := x.getRemoteServiceClient(to.GetHost(), int(to.GetPort()))
 	if err != nil {
 		return nil, err
 	}
 
-	senderPath := p.ActorPath()
+	senderPath := x.ActorPath()
 	senderAddress := senderPath.Address()
 
 	sender := &goaktpb.Address{
@@ -865,12 +906,12 @@ func (p *pid) RemoteAsk(ctx context.Context, to *goaktpb.Address, message proto.
 
 // RemoteBatchTell sends a batch of messages to a remote actor in a way fire-and-forget manner
 // Messages are processed one after the other in the order they are sent.
-func (p *pid) RemoteBatchTell(ctx context.Context, to *goaktpb.Address, messages ...proto.Message) error {
+func (x *pid) RemoteBatchTell(ctx context.Context, to *goaktpb.Address, messages ...proto.Message) error {
 	if len(messages) == 1 {
-		return p.RemoteTell(ctx, to, messages[0])
+		return x.RemoteTell(ctx, to, messages[0])
 	}
 
-	senderPath := p.ActorPath()
+	senderPath := x.ActorPath()
 	senderAddress := senderPath.Address()
 
 	sender := &goaktpb.Address{
@@ -896,7 +937,7 @@ func (p *pid) RemoteBatchTell(ctx context.Context, to *goaktpb.Address, messages
 		})
 	}
 
-	remoteService, err := p.getRemoteServiceClient(to.GetHost(), int(to.GetPort()))
+	remoteService, err := x.getRemoteServiceClient(to.GetHost(), int(to.GetPort()))
 	if err != nil {
 		return err
 	}
@@ -925,8 +966,8 @@ func (p *pid) RemoteBatchTell(ctx context.Context, to *goaktpb.Address, messages
 // RemoteBatchAsk sends a synchronous bunch of messages to a remote actor and expect responses in the same order as the messages.
 // Messages are processed one after the other in the order they are sent.
 // This can hinder performance if it is not properly used.
-func (p *pid) RemoteBatchAsk(ctx context.Context, to *goaktpb.Address, messages ...proto.Message) (responses []*anypb.Any, err error) {
-	senderPath := p.ActorPath()
+func (x *pid) RemoteBatchAsk(ctx context.Context, to *goaktpb.Address, messages ...proto.Message) (responses []*anypb.Any, err error) {
+	senderPath := x.ActorPath()
 	senderAddress := senderPath.Address()
 
 	sender := &goaktpb.Address{
@@ -952,7 +993,7 @@ func (p *pid) RemoteBatchAsk(ctx context.Context, to *goaktpb.Address, messages
 		})
 	}
 
-	remoteService, err := p.getRemoteServiceClient(to.GetHost(), int(to.GetPort()))
+	remoteService, err := x.getRemoteServiceClient(to.GetHost(), int(to.GetPort()))
 	if err != nil {
 		return nil, err
 	}
@@ -997,8 +1038,8 @@ func (p *pid) RemoteBatchAsk(ctx context.Context, to *goaktpb.Address, messages
 }
 
 // RemoteStop stops an actor on a remote node
-func (p *pid) RemoteStop(ctx context.Context, host string, port int, name string) error {
-	remoteService, err := p.getRemoteServiceClient(host, port)
+func (x *pid) RemoteStop(ctx context.Context, host string, port int, name string) error {
+	remoteService, err := x.getRemoteServiceClient(host, port)
 	if err != nil {
 		return err
 	}
@@ -1021,8 +1062,8 @@ func (p *pid) RemoteStop(ctx context.Context, host string, port int, name string
 }
 
 // RemoteSpawn creates an actor on a remote node. The given actor needs to be registered on the remote node using the Register method of ActorSystem
-func (p *pid) RemoteSpawn(ctx context.Context, host string, port int, name, actorType string) error {
-	remoteService, err := p.getRemoteServiceClient(host, port)
+func (x *pid) RemoteSpawn(ctx context.Context, host string, port int, name, actorType string) error {
+	remoteService, err := x.getRemoteServiceClient(host, port)
 	if err != nil {
 		return err
 	}
@@ -1051,8 +1092,8 @@ func (p *pid) RemoteSpawn(ctx context.Context, host string, port int, name, acto
 }
 
 // RemoteReSpawn restarts an actor on a remote node.
-func (p *pid) RemoteReSpawn(ctx context.Context, host string, port int, name string) error {
-	remoteService, err := p.getRemoteServiceClient(host, port)
+func (x *pid) RemoteReSpawn(ctx context.Context, host string, port int, name string) error {
+	remoteService, err := x.getRemoteServiceClient(host, port)
 	if err != nil {
 		return err
 	}
@@ -1077,59 +1118,59 @@ func (p *pid) RemoteReSpawn(ctx context.Context, host string, port int, name str
 // Shutdown gracefully shuts down the given actor
 // All current messages in the mailbox will be processed before the actor shutdown after a period of time
 // that can be configured. All child actors will be gracefully shutdown.
-func (p *pid) Shutdown(ctx context.Context) error {
-	spanCtx, span := p.tracer.Start(ctx, "Shutdown")
+func (x *pid) Shutdown(ctx context.Context) error {
+	spanCtx, span := x.tracer.Start(ctx, "Shutdown")
 	defer span.End()
 
-	p.stopLocker.Lock()
-	defer p.stopLocker.Unlock()
+	x.stopLocker.Lock()
+	defer x.stopLocker.Unlock()
 
-	p.logger.Info("Shutdown process has started...")
+	x.logger.Info("Shutdown process has started...")
 
-	if !p.isRunning.Load() {
-		p.logger.Infof("Actor=%s is offline. Maybe it has been passivated or stopped already", p.ActorPath().String())
+	if !x.isRunning.Load() {
+		x.logger.Infof("Actor=%s is offline. Maybe it has been passivated or stopped already", x.ActorPath().String())
 		span.SetStatus(codes.Ok, "Shutdown")
 		return nil
 	}
 
-	if p.passivateAfter.Load() > 0 {
-		p.logger.Debug("sending a signal to stop passivation listener....")
-		p.haltPassivationLnr <- types.Unit{}
+	if x.passivateAfter.Load() > 0 {
+		x.logger.Debug("sending a signal to stop passivation listener....")
+		x.haltPassivationLnr <- types.Unit{}
 	}
 
-	if err := p.doStop(spanCtx); err != nil {
-		p.logger.Errorf("failed to stop actor=(%s)", p.ActorPath().String())
+	if err := x.doStop(spanCtx); err != nil {
+		x.logger.Errorf("failed to stop actor=(%s)", x.ActorPath().String())
 		return err
 	}
 
-	if p.eventsStream != nil {
-		p.eventsStream.Publish(eventsTopic, &goaktpb.ActorStopped{
-			Address:   p.ActorPath().RemoteAddress(),
+	if x.eventsStream != nil {
+		x.eventsStream.Publish(eventsTopic, &goaktpb.ActorStopped{
+			Address:   x.ActorPath().RemoteAddress(),
 			StoppedAt: timestamppb.Now(),
 		})
 	}
 
-	p.logger.Infof("Actor=%s successfully shutdown", p.ActorPath().String())
+	x.logger.Infof("Actor=%s successfully shutdown", x.ActorPath().String())
 	span.SetStatus(codes.Ok, "Shutdown")
 	return nil
 }
 
 // Watch a pid for errors, and send on the returned channel if an error occurred
-func (p *pid) Watch(cid PID) {
+func (x *pid) Watch(cid PID) {
 	w := &watcher{
-		WatcherID: p,
+		WatcherID: x,
 		ErrChan:   make(chan error, 1),
 		Done:      make(chan types.Unit, 1),
 	}
 	cid.watchers().Append(w)
-	go p.supervise(cid, w)
+	go x.supervise(cid, w)
 }
 
 // UnWatch stops watching a given actor
-func (p *pid) UnWatch(pid PID) {
+func (x *pid) UnWatch(pid PID) {
 	for item := range pid.watchers().Iter() {
 		w := item.Value
-		if w.WatcherID.ActorPath().Equals(p.ActorPath()) {
+		if w.WatcherID.ActorPath().Equals(x.ActorPath()) {
 			w.Done <- types.Unit{}
 			pid.watchers().Delete(item.Index)
 			break
@@ -1138,54 +1179,54 @@ func (p *pid) UnWatch(pid PID) {
 }
 
 // Watchers return the list of watchersList
-func (p *pid) watchers() *slices.ConcurrentSlice[*watcher] {
-	return p.watchersList
+func (x *pid) watchers() *slices.ConcurrentSlice[*watcher] {
+	return x.watchersList
 }
 
 // doReceive pushes a given message to the actor receiveContextBuffer
-func (p *pid) doReceive(ctx ReceiveContext) {
-	p.rwLocker.Lock()
-	mailbox := p.mailbox
-	p.rwLocker.Unlock()
+func (x *pid) doReceive(ctx ReceiveContext) {
+	x.rwLocker.Lock()
+	mailbox := x.mailbox
+	x.rwLocker.Unlock()
 
-	p.lastProcessingTime.Store(time.Now())
+	x.lastProcessingTime.Store(time.Now())
 	if err := mailbox.Push(ctx); err != nil {
-		p.logger.Warn(err)
-		p.handleError(ctx, err)
+		x.logger.Warn(err)
+		x.handleError(ctx, err)
 	}
 }
 
 // init initializes the given actor and init processing messages
 // when the initialization failed the actor will not be started
-func (p *pid) init(ctx context.Context) error {
-	spanCtx, span := p.tracer.Start(ctx, "Init")
+func (x *pid) init(ctx context.Context) error {
+	spanCtx, span := x.tracer.Start(ctx, "Init")
 	defer span.End()
 
-	p.logger.Info("Initialization process has started...")
+	x.logger.Info("Initialization process has started...")
 
-	cancelCtx, cancel := context.WithTimeout(spanCtx, p.initTimeout.Load())
+	cancelCtx, cancel := context.WithTimeout(spanCtx, x.initTimeout.Load())
 	defer cancel()
 
 	// create a new retrier that will try a maximum of `initMaxRetries` times, with
 	// an initial delay of 100 ms and a maximum delay of 1 second
-	retrier := retry.NewRetrier(int(p.initMaxRetries.Load()), 100*time.Millisecond, time.Second)
-	if err := retrier.RunContext(cancelCtx, p.Actor.PreStart); err != nil {
+	retrier := retry.NewRetrier(int(x.initMaxRetries.Load()), 100*time.Millisecond, time.Second)
+	if err := retrier.RunContext(cancelCtx, x.Actor.PreStart); err != nil {
 		e := ErrInitFailure(err)
 		span.SetStatus(codes.Error, "Init")
 		span.RecordError(e)
 		return e
 	}
 
-	p.rwLocker.Lock()
-	p.isRunning.Store(true)
-	p.rwLocker.Unlock()
+	x.rwLocker.Lock()
+	x.isRunning.Store(true)
+	x.rwLocker.Unlock()
 
-	p.logger.Info("Initialization process successfully completed.")
+	x.logger.Info("Initialization process successfully completed.")
 	span.SetStatus(codes.Ok, "Init")
 
-	if p.eventsStream != nil {
-		p.eventsStream.Publish(eventsTopic, &goaktpb.ActorStarted{
-			Address:   p.ActorPath().RemoteAddress(),
+	if x.eventsStream != nil {
+		x.eventsStream.Publish(eventsTopic, &goaktpb.ActorStarted{
+			Address:   x.ActorPath().RemoteAddress(),
 			StartedAt: timestamppb.Now(),
 		})
 	}
@@ -1194,63 +1235,63 @@ func (p *pid) init(ctx context.Context) error {
 }
 
 // reset re-initializes the actor PID
-func (p *pid) reset() {
-	p.lastProcessingTime.Store(time.Time{})
-	p.passivateAfter.Store(DefaultPassivationTimeout)
-	p.replyTimeout.Store(DefaultReplyTimeout)
-	p.shutdownTimeout.Store(DefaultShutdownTimeout)
-	p.initMaxRetries.Store(DefaultInitMaxRetries)
-	p.lastProcessingDuration.Store(0)
-	p.initTimeout.Store(DefaultInitTimeout)
-	p.children = newPIDMap(10)
-	p.watchersList = slices.NewConcurrentSlice[*watcher]()
-	p.telemetry = telemetry.New()
-	p.mailbox.Reset()
-	p.resetBehavior()
-	if p.metricEnabled.Load() {
-		if err := p.registerMetrics(); err != nil {
-			p.logger.Error(errors.Wrapf(err, "failed to register actor=%s metrics", p.ActorPath().String()))
+func (x *pid) reset() {
+	x.lastProcessingTime.Store(time.Time{})
+	x.passivateAfter.Store(DefaultPassivationTimeout)
+	x.replyTimeout.Store(DefaultReplyTimeout)
+	x.shutdownTimeout.Store(DefaultShutdownTimeout)
+	x.initMaxRetries.Store(DefaultInitMaxRetries)
+	x.lastProcessingDuration.Store(0)
+	x.initTimeout.Store(DefaultInitTimeout)
+	x.children = newPIDMap(10)
+	x.watchersList = slices.NewConcurrentSlice[*watcher]()
+	x.telemetry = telemetry.New()
+	x.mailbox.Reset()
+	x.resetBehavior()
+	if x.metricEnabled.Load() {
+		if err := x.registerMetrics(); err != nil {
+			x.logger.Error(errors.Wrapf(err, "failed to register actor=%s metrics", x.ActorPath().String()))
 		}
 	}
 }
 
-func (p *pid) freeWatchers(ctx context.Context) {
-	p.logger.Debug("freeing all watcher actors...")
+func (x *pid) freeWatchers(ctx context.Context) {
+	x.logger.Debug("freeing all watcher actors...")
 
-	watchers := p.watchers()
+	watchers := x.watchers()
 	if watchers.Len() > 0 {
 		for item := range watchers.Iter() {
 			watcher := item.Value
 			terminated := &goaktpb.Terminated{}
 			if watcher.WatcherID.IsRunning() {
 				// TODO: handle error and push to some system dead-letters queue
-				_ = p.Tell(ctx, watcher.WatcherID, terminated)
-				watcher.WatcherID.UnWatch(p)
-				watcher.WatcherID.removeChild(p)
+				_ = x.Tell(ctx, watcher.WatcherID, terminated)
+				watcher.WatcherID.UnWatch(x)
+				watcher.WatcherID.removeChild(x)
 			}
 		}
 	}
 }
 
-func (p *pid) freeChildren(ctx context.Context) {
-	p.logger.Debug("freeing all child actors...")
+func (x *pid) freeChildren(ctx context.Context) {
+	x.logger.Debug("freeing all child actors...")
 
-	for _, child := range p.Children() {
-		p.UnWatch(child)
-		p.children.delete(child.ActorPath())
+	for _, child := range x.Children() {
+		x.UnWatch(child)
+		x.children.delete(child.ActorPath())
 		if err := child.Shutdown(ctx); err != nil {
-			p.logger.Panic(err)
+			x.logger.Panic(err)
 		}
 	}
 }
 
 // receive handles every mail in the actor receiveContextBuffer
-func (p *pid) receive() {
+func (x *pid) receive() {
 	for {
 		select {
-		case <-p.shutdownSignal:
+		case <-x.shutdownSignal:
 			return
-		case received, ok := <-p.mailbox.Iterator():
+		case received, ok := <-x.mailbox.Iterator():
 			// break out of the loop when the channel is closed
 			if !ok {
 				return
@@ -1259,62 +1300,62 @@ func (p *pid) receive() {
 			switch received.Message().(type) {
 			case *goaktpb.PoisonPill:
 				// stop the actor
-				_ = p.Shutdown(received.Context())
+				_ = x.Shutdown(received.Context())
 			default:
-				p.handleReceived(received)
+				x.handleReceived(received)
 			}
 		}
 	}
 }
 
-func (p *pid) handleReceived(received ReceiveContext) {
+func (x *pid) handleReceived(received ReceiveContext) {
 	defer func() {
 		if r := recover(); r != nil {
 			err := fmt.Errorf("%s", r)
-			for item := range p.watchersList.Iter() {
+			for item := range x.watchersList.Iter() {
 				item.Value.ErrChan <- err
 			}
 		}
 	}()
 	// send the message to the current actor behavior
-	if behavior, ok := p.behaviorStack.Peek(); ok {
+	if behavior, ok := x.behaviorStack.Peek(); ok {
 		behavior(received)
 	}
 }
 
 // supervise watches for child actor's failure and act based upon the supervisory strategy
-func (p *pid) supervise(cid PID, watcher *watcher) {
+func (x *pid) supervise(cid PID, watcher *watcher) {
 	for {
 		select {
 		case <-watcher.Done:
-			p.logger.Debugf("stop watching cid=(%s)", cid.ActorPath().String())
+			x.logger.Debugf("stop watching cid=(%s)", cid.ActorPath().String())
 			return
 		case err := <-watcher.ErrChan:
-			p.logger.Errorf("child actor=(%s) is failing: Err=%v", cid.ActorPath().String(), err)
-			switch p.supervisorStrategy {
+			x.logger.Errorf("child actor=(%s) is failing: Err=%v", cid.ActorPath().String(), err)
+			switch x.supervisorStrategy {
 			case StopDirective:
-				p.UnWatch(cid)
-				p.children.delete(cid.ActorPath())
+				x.UnWatch(cid)
+				x.children.delete(cid.ActorPath())
 				if err := cid.Shutdown(context.Background()); err != nil {
 					// this can enter into some infinite loop if we panic
 					// since we are just shutting down the actor we can just log the error
 					// TODO: rethink properly about PostStop error handling
-					p.logger.Error(err)
+					x.logger.Error(err)
 				}
 			case RestartDirective:
-				p.UnWatch(cid)
+				x.UnWatch(cid)
 				if err := cid.Restart(context.Background()); err != nil {
-					p.logger.Panic(err)
+					x.logger.Panic(err)
 				}
-				p.Watch(cid)
+				x.Watch(cid)
 			default:
-				p.UnWatch(cid)
-				p.children.delete(cid.ActorPath())
+				x.UnWatch(cid)
+				x.children.delete(cid.ActorPath())
 				if err := cid.Shutdown(context.Background()); err != nil {
 					// this can enter into some infinite loop if we panic
 					// since we are just shutting down the actor we can just log the error
 					// TODO: rethink properly about PostStop error handling
-					p.logger.Error(err)
+					x.logger.Error(err)
 				}
 			}
 		}
@@ -1323,10 +1364,10 @@ func (p *pid) supervise(cid PID, watcher *watcher) {
 
 // passivationListener checks whether the actor is processing public or not.
 // when the actor is idle, it automatically shuts down to free resources
-func (p *pid) passivationListener() {
-	p.logger.Info("start the passivation listener...")
-	p.logger.Infof("passivation timeout is (%s)", p.passivateAfter.Load().String())
-	ticker := time.NewTicker(p.passivateAfter.Load())
+func (x *pid) passivationListener() {
+	x.logger.Info("start the passivation listener...")
+	x.logger.Infof("passivation timeout is (%s)", x.passivateAfter.Load().String())
+	ticker := time.NewTicker(x.passivateAfter.Load())
 	tickerStopSig := make(chan types.Unit, 1)
 
 	// start ticking
@@ -1334,12 +1375,12 @@ func (p *pid) passivationListener() {
 		for {
 			select {
 			case <-ticker.C:
-				idleTime := time.Since(p.lastProcessingTime.Load())
-				if idleTime >= p.passivateAfter.Load() {
+				idleTime := time.Since(x.lastProcessingTime.Load())
+				if idleTime >= x.passivateAfter.Load() {
 					tickerStopSig <- types.Unit{}
 					return
 				}
-			case <-p.haltPassivationLnr:
+			case <-x.haltPassivationLnr:
 				tickerStopSig <- types.Unit{}
 				return
 			}
@@ -1349,76 +1390,76 @@ func (p *pid) passivationListener() {
 	<-tickerStopSig
 	ticker.Stop()
 
-	if !p.IsRunning() {
-		p.logger.Infof("Actor=%s is offline. No need to passivate", p.ActorPath().String())
+	if !x.IsRunning() {
+		x.logger.Infof("Actor=%s is offline. No need to passivate", x.ActorPath().String())
 		return
 	}
 
-	p.stopLocker.Lock()
-	defer p.stopLocker.Unlock()
+	x.stopLocker.Lock()
+	defer x.stopLocker.Unlock()
 
-	p.logger.Infof("Passivation mode has been triggered for actor=%s...", p.ActorPath().String())
+	x.logger.Infof("Passivation mode has been triggered for actor=%s...", x.ActorPath().String())
 
 	ctx := context.Background()
 
-	if err := p.doStop(ctx); err != nil {
+	if err := x.doStop(ctx); err != nil {
 		// TODO: rethink properly about PostStop error handling
-		p.logger.Errorf("failed to passivate actor=(%s): reason=(%v)", p.ActorPath().String(), err)
+		x.logger.Errorf("failed to passivate actor=(%s): reason=(%v)", x.ActorPath().String(), err)
 		return
 	}
 
-	if p.eventsStream != nil {
+	if x.eventsStream != nil {
 		event := &goaktpb.ActorPassivated{
-			Address:      p.ActorPath().RemoteAddress(),
+			Address:      x.ActorPath().RemoteAddress(),
 			PassivatedAt: timestamppb.Now(),
 		}
-		p.eventsStream.Publish(eventsTopic, event)
+		x.eventsStream.Publish(eventsTopic, event)
 	}
 
-	p.logger.Infof("Actor=%s successfully passivated", p.ActorPath().String())
+	x.logger.Infof("Actor=%s successfully passivated", x.ActorPath().String())
 }
 
 // setBehavior is a utility function that helps set the actor behavior
-func (p *pid) setBehavior(behavior Behavior) {
-	p.rwLocker.Lock()
-	p.behaviorStack.Clear()
-	p.behaviorStack.Push(behavior)
-	p.rwLocker.Unlock()
+func (x *pid) setBehavior(behavior Behavior) {
+	x.rwLocker.Lock()
+	x.behaviorStack.Clear()
+	x.behaviorStack.Push(behavior)
+	x.rwLocker.Unlock()
 }
 
 // resetBehavior is a utility function resets the actor behavior
-func (p *pid) resetBehavior() {
-	p.rwLocker.Lock()
-	p.behaviorStack.Clear()
-	p.behaviorStack.Push(p.Receive)
-	p.rwLocker.Unlock()
+func (x *pid) resetBehavior() {
+	x.rwLocker.Lock()
+	x.behaviorStack.Clear()
+	x.behaviorStack.Push(x.Receive)
+	x.rwLocker.Unlock()
 }
 
 // setBehaviorStacked adds a behavior to the actor's behaviorStack
-func (p *pid) setBehaviorStacked(behavior Behavior) {
-	p.rwLocker.Lock()
-	p.behaviorStack.Push(behavior)
-	p.rwLocker.Unlock()
+func (x *pid) setBehaviorStacked(behavior Behavior) {
+	x.rwLocker.Lock()
+	x.behaviorStack.Push(behavior)
+	x.rwLocker.Unlock()
 }
 
 // unsetBehaviorStacked sets the actor's behavior to the previous behavior
 // prior to setBehaviorStacked is called
-func (p *pid) unsetBehaviorStacked() {
-	p.rwLocker.Lock()
-	p.behaviorStack.Pop()
-	p.rwLocker.Unlock()
+func (x *pid) unsetBehaviorStacked() {
+	x.rwLocker.Lock()
+	x.behaviorStack.Pop()
+	x.rwLocker.Unlock()
 }
 
 // doStop stops the actor
-func (p *pid) doStop(ctx context.Context) error {
-	spanCtx, span := p.tracer.Start(ctx, "doStop")
+func (x *pid) doStop(ctx context.Context) error {
+	spanCtx, span := x.tracer.Start(ctx, "doStop")
 	defer span.End()
 
-	p.isRunning.Store(false)
+	x.isRunning.Store(false)
 
 	// TODO: just signal stash processing done and ignore the messages or process them
-	for p.stashBuffer != nil && !p.stashBuffer.IsEmpty() {
-		if err := p.unstashAll(); err != nil {
+	for x.stashBuffer != nil && !x.stashBuffer.IsEmpty() {
+		if err := x.unstashAll(); err != nil {
 			span.SetStatus(codes.Error, "doStop")
 			span.RecordError(err)
 			return err
@@ -1429,14 +1470,14 @@ func (p *pid) doStop(ctx context.Context) error {
 	// init a ticker that run every 10 ms to make sure we process all messages in the
 	// mailbox.
 	ticker := time.NewTicker(10 * time.Millisecond)
-	timer := time.After(p.shutdownTimeout.Load())
+	timer := time.After(x.shutdownTimeout.Load())
 	tickerStopSig := make(chan types.Unit)
 	// start ticking
 	go func() {
 		for {
 			select {
 			case <-ticker.C:
-				if p.mailbox.IsEmpty() {
+				if x.mailbox.IsEmpty() {
 					close(tickerStopSig)
 					return
 				}
@@ -1448,19 +1489,19 @@ func (p *pid) doStop(ctx context.Context) error {
 	}()
 
 	<-tickerStopSig
-	p.shutdownSignal <- types.Unit{}
-	p.httpClient.CloseIdleConnections()
-	p.freeChildren(spanCtx)
-	p.freeWatchers(spanCtx)
-
-	p.logger.Infof("Shutdown process is on going for actor=%s...", p.ActorPath().String())
-	p.reset()
-	return p.Actor.PostStop(spanCtx)
+	x.shutdownSignal <- types.Unit{}
+	x.httpClient.CloseIdleConnections()
+	x.freeChildren(spanCtx)
+	x.freeWatchers(spanCtx)
+
+	x.logger.Infof("Shutdown process is on going for actor=%s...", x.ActorPath().String())
+	x.reset()
+	return x.Actor.PostStop(spanCtx)
 }
 
 // removeChild helps remove child actor
-func (p *pid) removeChild(pid PID) {
-	if !p.IsRunning() {
+func (x *pid) removeChild(pid PID) {
+	if !x.IsRunning() {
 		return
 	}
 
@@ -1469,17 +1510,17 @@ func (p *pid) removeChild(pid PID) {
 	}
 
 	path := pid.ActorPath()
-	if cid, ok := p.children.get(path); ok {
+	if cid, ok := x.children.get(path); ok {
 		if cid.IsRunning() {
 			return
 		}
-		p.children.delete(path)
+		x.children.delete(path)
 	}
 }
 
 // handleError handles the error during a message handling
-func (p *pid) handleError(receiveCtx ReceiveContext, err error) {
-	if p.eventsStream == nil {
+func (x *pid) handleError(receiveCtx ReceiveContext, err error) {
+	if x.eventsStream == nil {
 		return
 	}
 
@@ -1497,7 +1538,7 @@ func (p *pid) handleError(receiveCtx ReceiveContext, err error) {
 		senderAddr = receiveCtx.Sender().ActorPath().RemoteAddress()
 	}
 
-	receiver := p.actorPath.RemoteAddress()
+	receiver := x.actorPath.RemoteAddress()
 	deadletter := &goaktpb.Deadletter{
 		Sender:   senderAddr,
 		Receiver: receiver,
@@ -1506,21 +1547,21 @@ func (p *pid) handleError(receiveCtx ReceiveContext, err error) {
 		Reason:   err.Error(),
 	}
 
-	p.eventsStream.Publish(eventsTopic, deadletter)
+	x.eventsStream.Publish(eventsTopic, deadletter)
 }
 
 // registerMetrics register the PID metrics with OTel instrumentation.
-func (p *pid) registerMetrics() error {
-	meter := p.telemetry.Meter
+func (x *pid) registerMetrics() error {
+	meter := x.telemetry.Meter
 	metrics, err := metric.NewActorMetric(meter)
 	if err != nil {
 		return err
 	}
 
 	_, err = meter.RegisterCallback(func(_ context.Context, observer otelmetric.Observer) error {
-		observer.ObserveInt64(metrics.ChildrenCount(), p.childrenCount.Load())
-		observer.ObserveInt64(metrics.StashCount(), int64(p.StashSize()))
-		observer.ObserveInt64(metrics.RestartCount(), p.restartCount.Load())
+		observer.ObserveInt64(metrics.ChildrenCount(), x.childrenCount.Load())
+		observer.ObserveInt64(metrics.StashCount(), int64(x.StashSize()))
+		observer.ObserveInt64(metrics.RestartCount(), x.restartCount.Load())
 		return nil
 	}, metrics.ChildrenCount(),
 		metrics.StashCount(),
@@ -1530,48 +1571,102 @@ func (p *pid) registerMetrics() error {
 }
 
 // getConnectionOptions returns the gRPC client connections options
-func (p *pid) getConnectionOptions() ([]connect.ClientOption, error) {
+func (x *pid) getConnectionOptions() ([]connect.ClientOption, error) {
 	var interceptor *otelconnect.Interceptor
 	var err error
-	if p.metricEnabled.Load() || p.traceEnabled.Load() {
+	if x.metricEnabled.Load() || x.traceEnabled.Load() {
 		interceptor, err = otelconnect.NewInterceptor(
-			otelconnect.WithTracerProvider(p.telemetry.TracerProvider),
-			otelconnect.WithMeterProvider(p.telemetry.MeterProvider))
+			otelconnect.WithTracerProvider(x.telemetry.TracerProvider),
+			otelconnect.WithMeterProvider(x.telemetry.MeterProvider))
 		if err != nil {
 			return nil, errors.Wrap(err, "failed to initialize observability feature")
 		}
 	}
 
-	var clientConnectionOptions []connect.ClientOption
+	var clientOptions []connect.ClientOption
 	if interceptor != nil {
-		clientConnectionOptions = append(clientConnectionOptions, connect.WithInterceptors(interceptor))
+		clientOptions = append(clientOptions, connect.WithInterceptors(interceptor))
 	}
-	return clientConnectionOptions, err
+	return clientOptions, err
 }
 
 // getRemoteServiceClient returns an instance of the Remote Service client
-func (p *pid) getRemoteServiceClient(host string, port int) (internalpbconnect.RemotingServiceClient, error) {
-	clientConnectionOptions, err := p.getConnectionOptions()
+func (x *pid) getRemoteServiceClient(host string, port int) (internalpbconnect.RemotingServiceClient, error) {
+	clientConnectionOptions, err := x.getConnectionOptions()
 	if err != nil {
 		return nil, err
 	}
 
 	return internalpbconnect.NewRemotingServiceClient(
-		p.httpClient,
+		x.httpClient,
 		http.URL(host, port),
 		clientConnectionOptions...,
 	), nil
 }
 
-func (p *pid) getLastProcessingTime() time.Time {
-	p.processingTimeLocker.Lock()
-	processingTime := p.lastProcessingTime.Load()
-	p.processingTimeLocker.Unlock()
+// getLastProcessingTime returns the last processing time
+func (x *pid) getLastProcessingTime() time.Time {
+	x.processingTimeLocker.Lock()
+	processingTime := x.lastProcessingTime.Load()
+	x.processingTimeLocker.Unlock()
 	return processingTime
 }
 
-func (p *pid) setLastProcessingDuration(d time.Duration) {
-	p.processingTimeLocker.Lock()
-	p.lastProcessingDuration.Store(d)
-	p.processingTimeLocker.Unlock()
+// setLastProcessingDuration sets the last processing duration
+func (x *pid) setLastProcessingDuration(d time.Duration) {
+	x.processingTimeLocker.Lock()
+	x.lastProcessingDuration.Store(d)
+	x.processingTimeLocker.Unlock()
+}
+
+// handleCompletion processes a long-running task and pipe the result to
+// the completion receiver
+func (x *pid) handleCompletion(ctx context.Context, completion *completion) {
+	ctx, span := x.tracer.Start(ctx, "TaskCompletion")
+	defer span.End()
+
+	// defensive programming
+	if completion == nil ||
+		completion.Receiver == nil ||
+		completion.Receiver == NoSender ||
+		completion.Task == nil {
+		x.logger.Error(ErrUndefinedTask)
+		span.RecordError(ErrUndefinedTask)
+		span.SetStatus(codes.Error, "TaskCompletion")
+		return
+	}
+
+	// wrap the provided completion task into a future that can help run the task
+	f := future.NewWithContext(ctx, completion.Task)
+	var wg sync.WaitGroup
+	var result *future.Result
+
+	// wait for a potential result or timeout
+	wg.Add(1)
+	go func() {
+		result = f.Result()
+		wg.Done()
+	}()
+	wg.Wait()
+
+	// logger the error when the task returns an error
+	if err := result.Failure(); err != nil {
+		x.logger.Error(err)
+		span.RecordError(err)
+		span.SetStatus(codes.Error, "TaskCompletion")
+		return
+	}
+
+	// make sure that the receiver is still alive
+	to := completion.Receiver
+	if !to.IsRunning() {
+		x.logger.Errorf("unable to pipe message to actor=(%s): not running", to.ActorPath().String())
+		span.RecordError(ErrDead)
+		span.SetStatus(codes.Error, "TaskCompletion")
+		return
+	}
+
+	messageContext := newReceiveContext(ctx, x, to, result.Success(), true)
+	to.doReceive(messageContext)
+	span.SetStatus(codes.Ok, "TaskCompletion")
 }
diff --git a/actors/pid_map.go b/actors/pid_map.go
index 3bd32b9..0086d8e 100644
--- a/actors/pid_map.go
+++ b/actors/pid_map.go
@@ -28,8 +28,9 @@ import (
 	"reflect"
 	"sync"
 
-	"github.com/tochemey/goakt/v2/internal/types"
 	"go.uber.org/atomic"
+
+	"github.com/tochemey/goakt/v2/internal/types"
 )
 
 type prop struct {
diff --git a/actors/pid_option.go b/actors/pid_option.go
index cb917e7..8965317 100644
--- a/actors/pid_option.go
+++ b/actors/pid_option.go
@@ -42,9 +42,9 @@ func withPassivationAfter(duration time.Duration) pidOption {
 	}
 }
 
-// withSendReplyTimeout sets how long in seconds an actor should reply a command
+// withReplyTimeout sets how long in seconds an actor should reply a command
 // in a receive-reply pattern
-func withSendReplyTimeout(timeout time.Duration) pidOption {
+func withReplyTimeout(timeout time.Duration) pidOption {
 	return func(pid *pid) {
 		pid.replyTimeout.Store(timeout)
 	}
diff --git a/actors/stash.go b/actors/stash.go
index 778670f..78d2751 100644
--- a/actors/stash.go
+++ b/actors/stash.go
@@ -25,46 +25,46 @@
 package actors
 
 // stash adds the current message to the stash buffer
-func (p *pid) stash(ctx ReceiveContext) error {
-	if p.stashBuffer == nil {
+func (x *pid) stash(ctx ReceiveContext) error {
+	if x.stashBuffer == nil {
 		return ErrStashBufferNotSet
 	}
 
-	p.stashLocker.Lock()
-	defer p.stashLocker.Unlock()
+	x.stashLocker.Lock()
+	defer x.stashLocker.Unlock()
 
-	return p.stashBuffer.Push(ctx)
+	return x.stashBuffer.Push(ctx)
 }
 
 // unstash unstashes the oldest message in the stash and prepends to the mailbox
-func (p *pid) unstash() error {
-	if p.stashBuffer == nil {
+func (x *pid) unstash() error {
+	if x.stashBuffer == nil {
 		return ErrStashBufferNotSet
 	}
 
-	p.stashLocker.Lock()
-	defer p.stashLocker.Unlock()
+	x.stashLocker.Lock()
+	defer x.stashLocker.Unlock()
 
-	if !p.stashBuffer.IsEmpty() {
-		received, _ := p.stashBuffer.Pop()
-		p.doReceive(received)
+	if !x.stashBuffer.IsEmpty() {
+		received, _ := x.stashBuffer.Pop()
+		x.doReceive(received)
 	}
 	return nil
 }
 
 // unstashAll unstashes all messages from the stash buffer and prepends in the mailbox
 // (it keeps the messages in the same order as received, unstashing older messages before newer).
-func (p *pid) unstashAll() error {
-	if p.stashBuffer == nil {
+func (x *pid) unstashAll() error {
+	if x.stashBuffer == nil {
 		return ErrStashBufferNotSet
 	}
 
-	p.stashLocker.Lock()
-	defer p.stashLocker.Unlock()
+	x.stashLocker.Lock()
+	defer x.stashLocker.Unlock()
 
-	for !p.stashBuffer.IsEmpty() {
-		received, _ := p.stashBuffer.Pop()
-		p.doReceive(received)
+	for !x.stashBuffer.IsEmpty() {
+		received, _ := x.stashBuffer.Pop()
+		x.doReceive(received)
 	}
 
 	return nil
diff --git a/actors/constants.go b/actors/types.go
similarity index 100%
rename from actors/constants.go
rename to actors/types.go
diff --git a/future/future.go b/future/future.go
new file mode 100644
index 0000000..f53b2a2
--- /dev/null
+++ b/future/future.go
@@ -0,0 +1,155 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2022-2024 Tochemey
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+// Package future
+// Most of the code here has been copied from (https://github.com/Workiva/go-datastructures) with a slight modification
+package future
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"google.golang.org/protobuf/proto"
+)
+
+// ErrTimeout is returned when the future has timed out
+var ErrTimeout = func(duration time.Duration) error { return fmt.Errorf(`timeout after %f seconds`, duration.Seconds()) }
+
+// Task defines the successful outcome of a long-running task
+type Task <-chan proto.Message
+
+// Result defines the future result.
+// It holds both the success and error result
+type Result struct {
+	success proto.Message
+	failure error
+}
+
+// Success returns the future success result
+func (x *Result) Success() proto.Message {
+	return x.success
+}
+
+// Failure returns the future error result
+func (x *Result) Failure() error {
+	return x.failure
+}
+
+// Future represents an object that can be used to perform asynchronous
+// tasks.  The constructor of the future will complete it, and listeners
+// will block on Result until a result is received.  This is different
+// from a channel in that the future is only completed once, and anyone
+// listening on the future will get the result, regardless of the number
+// of listeners.
+type Future struct {
+	triggered bool // because result can technically be nil and still be valid
+	result    *Result
+	lock      sync.Mutex
+	wg        sync.WaitGroup
+}
+
+// New creates an instance of Future that will time out when the timeout is hit
+func New(task Task, timeout time.Duration) *Future {
+	f := new(Future)
+	f.wg.Add(1)
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go waitTimeout(f, task, timeout, &wg)
+	wg.Wait()
+	return f
+}
+
+// NewWithContext creates an instance of Future with a context.
+// The future will time out when the given context is canceled before the response is received
+func NewWithContext(ctx context.Context, task Task) *Future {
+	f := new(Future)
+	f.wg.Add(1)
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go wait(ctx, f, task, &wg)
+	wg.Wait()
+	return f
+}
+
+// Result will immediately fetch the result if it exists
+// or wait on the result until it is ready.
+func (f *Future) Result() *Result {
+	f.lock.Lock()
+	if f.triggered {
+		f.lock.Unlock()
+		return f.result
+	}
+	f.lock.Unlock()
+
+	f.wg.Wait()
+	return f.result
+}
+
+// HasResult will return true iff the result exists
+func (f *Future) HasResult() bool {
+	f.lock.Lock()
+	hasResult := f.triggered
+	f.lock.Unlock()
+	return hasResult
+}
+
+// setResult sets the completed future result
+func (f *Future) setResult(success proto.Message, err error) {
+	f.lock.Lock()
+	f.triggered = true
+	f.result = &Result{
+		success: success,
+		failure: err,
+	}
+	f.lock.Unlock()
+	f.wg.Done()
+}
+
+// waitTimeout waits for the result or times out
+func waitTimeout(f *Future, ch Task, timeout time.Duration, wg *sync.WaitGroup) {
+	wg.Done()
+	t := time.NewTimer(timeout)
+	select {
+	case success := <-ch:
+		f.setResult(success, nil)
+		t.Stop()
+	case <-t.C:
+		var _nil proto.Message
+		f.setResult(_nil, ErrTimeout(timeout))
+	}
+}
+
+// wait waits for the result until the provided context is canceled
+func wait(ctx context.Context, f *Future, ch Task, wg *sync.WaitGroup) {
+	wg.Done()
+	select {
+	case success := <-ch:
+		f.setResult(success, nil)
+	case <-ctx.Done():
+		var _nil proto.Message
+		f.setResult(_nil, ctx.Err())
+	}
+}
diff --git a/go.mod b/go.mod
index 5d93d66..cd05227 100644
--- a/go.mod
+++ b/go.mod
@@ -12,7 +12,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/grandcat/zeroconf v1.0.0
 	github.com/nats-io/nats-server/v2 v2.10.16
-	github.com/nats-io/nats.go v1.35.0
+	github.com/nats-io/nats.go v1.36.0
 	github.com/pkg/errors v0.9.1
 	github.com/redis/go-redis/v9 v9.5.3
 	github.com/reugn/go-quartz v0.12.0
diff --git a/go.sum b/go.sum
index 7e0159a..c3a10e8 100644
--- a/go.sum
+++ b/go.sum
@@ -188,8 +188,8 @@ github.com/nats-io/jwt/v2 v2.5.7 h1:j5lH1fUXCnJnY8SsQeB/a/z9Azgu2bYIDvtPVNdxe2c=
 github.com/nats-io/jwt/v2 v2.5.7/go.mod h1:ZdWS1nZa6WMZfFwwgpEaqBV8EPGVgOTDHN/wTbz0Y5A=
 github.com/nats-io/nats-server/v2 v2.10.16 h1:2jXaiydp5oB/nAx/Ytf9fdCi9QN6ItIc9eehX8kwVV0=
 github.com/nats-io/nats-server/v2 v2.10.16/go.mod h1:Pksi38H2+6xLe1vQx0/EA4bzetM0NqyIHcIbmgXSkIU=
-github.com/nats-io/nats.go v1.35.0 h1:XFNqNM7v5B+MQMKqVGAyHwYhyKb48jrenXNxIU20ULk=
-github.com/nats-io/nats.go v1.35.0/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
+github.com/nats-io/nats.go v1.36.0 h1:suEUPuWzTSse/XhESwqLxXGuj8vGRuPRoG7MoRN/qyU=
+github.com/nats-io/nats.go v1.36.0/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
 github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=
 github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=


Test patch:
diff --git a/actors/actor_system_test.go b/actors/actor_system_test.go
index 5ae295d..3fef36b 100644
--- a/actors/actor_system_test.go
+++ b/actors/actor_system_test.go
@@ -382,7 +382,7 @@ func TestActorSystem(t *testing.T) {
 		assert.NotNil(t, actorRef)
 
 		// send a message to the actor
-		reply, err := Ask(ctx, actorRef, new(testpb.TestReply), receivingTimeout)
+		reply, err := Ask(ctx, actorRef, new(testpb.TestReply), replyTimeout)
 		require.NoError(t, err)
 		require.NotNil(t, reply)
 		expected := new(testpb.Reply)
@@ -462,7 +462,7 @@ func TestActorSystem(t *testing.T) {
 		assert.NotNil(t, actorRef)
 
 		// send a message to the actor
-		reply, err := Ask(ctx, actorRef, new(testpb.TestReply), receivingTimeout)
+		reply, err := Ask(ctx, actorRef, new(testpb.TestReply), replyTimeout)
 		require.NoError(t, err)
 		require.NotNil(t, reply)
 		expected := new(testpb.Reply)
@@ -517,7 +517,7 @@ func TestActorSystem(t *testing.T) {
 		assert.NotNil(t, actorRef)
 
 		// send a message to the actor
-		reply, err := Ask(ctx, actorRef, new(testpb.TestReply), receivingTimeout)
+		reply, err := Ask(ctx, actorRef, new(testpb.TestReply), replyTimeout)
 		require.NoError(t, err)
 		require.NotNil(t, reply)
 		expected := new(testpb.Reply)
diff --git a/actors/actor_test.go b/actors/actor_test.go
index d67fe19..3ccb0b9 100644
--- a/actors/actor_test.go
+++ b/actors/actor_test.go
@@ -221,9 +221,20 @@ func (x *userActor) DebitAccount(ctx ReceiveContext) {
 	}
 }
 
-type exchanger struct{}
+type exchanger struct {
+	messageCounter *atomic.Int64
+	logger         log.Logger
+	id             string
+}
+
+// Counter is only here for testing purpose
+func (e *exchanger) Counter() int64 {
+	return e.messageCounter.Load()
+}
 
 func (e *exchanger) PreStart(context.Context) error {
+	e.messageCounter = atomic.NewInt64(0)
+	e.logger = log.DefaultLogger
 	return nil
 }
 
@@ -231,11 +242,19 @@ func (e *exchanger) Receive(ctx ReceiveContext) {
 	message := ctx.Message()
 	switch message.(type) {
 	case *goaktpb.PostStart:
+		e.id = ctx.Self().ID()
 	case *testspb.TestSend:
+		e.messageCounter.Inc()
 		_ = ctx.Self().Tell(ctx.Context(), ctx.Sender(), new(testspb.TestSend))
+	case *testspb.TaskComplete:
+		e.messageCounter.Inc()
+		e.logger.Debugf("actor=(%s) has processed so far: %d", e.id, e.messageCounter.Load())
 	case *testspb.TestReply:
+		e.messageCounter.Inc()
+		e.logger.Debugf("actor=(%s) has processed so far: %d", e.id, e.messageCounter.Load())
 		ctx.Response(new(testspb.Reply))
 	case *testspb.TestRemoteSend:
+		e.messageCounter.Inc()
 		_ = ctx.Self().RemoteTell(context.Background(), ctx.RemoteSender(), new(testspb.TestBye))
 	case *testspb.TestBye:
 		_ = ctx.Self().Shutdown(ctx.Context())
diff --git a/actors/api_test.go b/actors/api_test.go
index 006a3c5..3ec3c91 100644
--- a/actors/api_test.go
+++ b/actors/api_test.go
@@ -72,7 +72,7 @@ func TestAsk(t *testing.T) {
 		// 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)
+		reply, err := Ask(ctx, actorRef, message, replyTimeout)
 		// perform some assertions
 		require.NoError(t, err)
 		assert.NotNil(t, reply)
@@ -113,7 +113,7 @@ func TestAsk(t *testing.T) {
 		// 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)
+		reply, err := Ask(ctx, actorRef, message, replyTimeout)
 		// perform some assertions
 		require.Error(t, err)
 		assert.EqualError(t, err, ErrDead.Error())
@@ -129,7 +129,7 @@ func TestAsk(t *testing.T) {
 		// create the actor system
 		sys, err := NewActorSystem("test",
 			WithLogger(logger),
-			WithReplyTimeout(receivingTimeout),
+			WithReplyTimeout(replyTimeout),
 			WithPassivationDisabled())
 		// assert there are no error
 		require.NoError(t, err)
@@ -150,7 +150,7 @@ func TestAsk(t *testing.T) {
 		// create a message to send to the test actor
 		message := new(testpb.TestTimeout)
 		// send the message to the actor
-		reply, err := Ask(ctx, actorRef, message, receivingTimeout)
+		reply, err := Ask(ctx, actorRef, message, replyTimeout)
 		// perform some assertions
 		require.Error(t, err)
 		assert.EqualError(t, err, ErrRequestTimeout.Error())
@@ -190,7 +190,7 @@ func TestAsk(t *testing.T) {
 			Message: &anypb.Any{},
 		}
 		// send the message to the actor
-		reply, err := Ask(ctx, actorRef, message, receivingTimeout)
+		reply, err := Ask(ctx, actorRef, message, replyTimeout)
 		// perform some assertions
 		require.Error(t, err)
 		assert.Nil(t, reply)
@@ -224,7 +224,7 @@ func TestAsk(t *testing.T) {
 
 		// create a message to send to the test actor
 		// send the message to the actor
-		replies, err := BatchAsk(ctx, actorRef, receivingTimeout, new(testpb.TestReply), new(testpb.TestReply))
+		replies, err := BatchAsk(ctx, actorRef, replyTimeout, new(testpb.TestReply), new(testpb.TestReply))
 		// perform some assertions
 		require.NoError(t, err)
 		assert.NotNil(t, replies)
@@ -246,7 +246,7 @@ func TestAsk(t *testing.T) {
 		// create the actor system
 		sys, err := NewActorSystem("test",
 			WithLogger(logger),
-			WithReplyTimeout(receivingTimeout),
+			WithReplyTimeout(replyTimeout),
 			WithPassivationDisabled())
 		// assert there are no error
 		require.NoError(t, err)
@@ -266,7 +266,7 @@ func TestAsk(t *testing.T) {
 
 		// create a message to send to the test actor
 		// send the message to the actor
-		replies, err := BatchAsk(ctx, actorRef, receivingTimeout, new(testpb.TestTimeout), new(testpb.TestReply))
+		replies, err := BatchAsk(ctx, actorRef, replyTimeout, new(testpb.TestTimeout), new(testpb.TestReply))
 		// perform some assertions
 		require.Error(t, err)
 		require.EqualError(t, err, ErrRequestTimeout.Error())
@@ -286,7 +286,7 @@ func TestAsk(t *testing.T) {
 		// create the actor system
 		sys, err := NewActorSystem("test",
 			WithLogger(logger),
-			WithReplyTimeout(receivingTimeout),
+			WithReplyTimeout(replyTimeout),
 			WithPassivationDisabled())
 		// assert there are no error
 		require.NoError(t, err)
@@ -309,7 +309,7 @@ func TestAsk(t *testing.T) {
 
 		// create a message to send to the test actor
 		// send the message to the actor
-		replies, err := BatchAsk(ctx, actorRef, receivingTimeout, new(testpb.TestTimeout), new(testpb.TestReply))
+		replies, err := BatchAsk(ctx, actorRef, replyTimeout, new(testpb.TestTimeout), new(testpb.TestReply))
 		// perform some assertions
 		require.Error(t, err)
 		assert.Empty(t, replies)
diff --git a/actors/context_test.go b/actors/context_test.go
index be58ffb..920f4be 100644
--- a/actors/context_test.go
+++ b/actors/context_test.go
@@ -41,6 +41,7 @@ import (
 	"github.com/tochemey/goakt/v2/internal/eventstream"
 	"github.com/tochemey/goakt/v2/log"
 	testpb "github.com/tochemey/goakt/v2/test/data/testpb"
+	testspb "github.com/tochemey/goakt/v2/test/data/testpb"
 )
 
 func TestReceiveContext(t *testing.T) {
@@ -61,35 +62,35 @@ func TestReceiveContext(t *testing.T) {
 
 		// send Login
 		var expected proto.Message
-		success, err := Ask(ctx, pid, new(testpb.TestLogin), receivingTimeout)
+		success, err := Ask(ctx, pid, new(testpb.TestLogin), replyTimeout)
 		require.NoError(t, err)
 		require.NotNil(t, success)
 		expected = &testpb.TestLoginSuccess{}
 		require.True(t, proto.Equal(expected, success))
 
 		// ask for readiness
-		ready, err := Ask(ctx, pid, new(testpb.TestReadiness), receivingTimeout)
+		ready, err := Ask(ctx, pid, new(testpb.TestReadiness), replyTimeout)
 		require.NoError(t, err)
 		require.NotNil(t, ready)
 		expected = &testpb.TestReady{}
 		require.True(t, proto.Equal(expected, ready))
 
 		// send a message to create account
-		created, err := Ask(ctx, pid, new(testpb.CreateAccount), receivingTimeout)
+		created, err := Ask(ctx, pid, new(testpb.CreateAccount), replyTimeout)
 		require.NoError(t, err)
 		require.NotNil(t, created)
 		expected = &testpb.AccountCreated{}
 		require.True(t, proto.Equal(expected, created))
 
 		// credit account
-		credited, err := Ask(ctx, pid, new(testpb.CreditAccount), receivingTimeout)
+		credited, err := Ask(ctx, pid, new(testpb.CreditAccount), replyTimeout)
 		require.NoError(t, err)
 		require.NotNil(t, credited)
 		expected = &testpb.AccountCredited{}
 		require.True(t, proto.Equal(expected, credited))
 
 		// debit account
-		debited, err := Ask(ctx, pid, new(testpb.DebitAccount), receivingTimeout)
+		debited, err := Ask(ctx, pid, new(testpb.DebitAccount), replyTimeout)
 		require.NoError(t, err)
 		require.NotNil(t, debited)
 		expected = &testpb.AccountDebited{}
@@ -687,7 +688,7 @@ func TestReceiveContext(t *testing.T) {
 			newSupervisor(),
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -725,7 +726,7 @@ func TestReceiveContext(t *testing.T) {
 			newSupervisor(),
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -757,7 +758,7 @@ func TestReceiveContext(t *testing.T) {
 			newSupervisor(),
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -798,7 +799,7 @@ func TestReceiveContext(t *testing.T) {
 			newSupervisor(),
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -839,7 +840,7 @@ func TestReceiveContext(t *testing.T) {
 			newSupervisor(),
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -880,7 +881,7 @@ func TestReceiveContext(t *testing.T) {
 			newSupervisor(),
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -923,7 +924,7 @@ func TestReceiveContext(t *testing.T) {
 			newSupervisor(),
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -957,7 +958,7 @@ func TestReceiveContext(t *testing.T) {
 			newSupervisor(),
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -997,7 +998,7 @@ func TestReceiveContext(t *testing.T) {
 			newSupervisor(),
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -1018,7 +1019,7 @@ func TestReceiveContext(t *testing.T) {
 			newSupervisor(),
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 
@@ -1041,7 +1042,7 @@ func TestReceiveContext(t *testing.T) {
 			newSupervisor(),
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -1291,7 +1292,6 @@ func TestReceiveContext(t *testing.T) {
 			context.Shutdown()
 		})
 	})
-
 	t.Run("With Unhandled with system messages", func(t *testing.T) {
 		ctx := context.TODO()
 		// create the deadletter stream
@@ -1352,7 +1352,6 @@ func TestReceiveContext(t *testing.T) {
 			context.Shutdown()
 		})
 	})
-
 	t.Run("With successful BatchTell", func(t *testing.T) {
 		ctx := context.TODO()
 		// create a Ping actor
@@ -1925,4 +1924,115 @@ func TestReceiveContext(t *testing.T) {
 			assert.NoError(t, sys.Stop(ctx))
 		})
 	})
+	t.Run("With successful PipeTo", func(t *testing.T) {
+		askTimeout := time.Minute
+		ctx := context.TODO()
+
+		opts := []pidOption{
+			withInitMaxRetries(1),
+			withReplyTimeout(askTimeout),
+			withPassivationDisabled(),
+			withCustomLogger(log.DefaultLogger),
+		}
+
+		// create actor1
+		actor1 := &exchanger{}
+		actorPath1 := NewPath("Exchange1", NewAddress("sys", "host", 1))
+		pid1, err := newPID(ctx, actorPath1, actor1, opts...)
+		require.NoError(t, err)
+		require.NotNil(t, pid1)
+
+		// create actor2
+		actor2 := &exchanger{}
+		actorPath2 := NewPath("Exchange2", NewAddress("sys", "host", 1))
+		pid2, err := newPID(ctx, actorPath2, actor2, opts...)
+		require.NoError(t, err)
+		require.NotNil(t, pid2)
+
+		// zero message received by both actors
+		require.Zero(t, actor1.Counter())
+		require.Zero(t, actor2.Counter())
+
+		// create an instance of receive context
+		messageContext := &receiveContext{
+			ctx:            ctx,
+			message:        new(testpb.TaskComplete),
+			sender:         NoSender,
+			recipient:      pid1,
+			mu:             sync.Mutex{},
+			isAsyncMessage: true,
+		}
+
+		task := make(chan proto.Message)
+		messageContext.PipeTo(pid2, task)
+
+		var wg sync.WaitGroup
+		wg.Add(1)
+		go func() {
+			// Wait for some time and during that period send some messages to the actor
+			// send three messages while waiting for the future to completed
+			_, _ = Ask(ctx, pid1, new(testpb.TestReply), askTimeout)
+			_, _ = Ask(ctx, pid1, new(testpb.TestReply), askTimeout)
+			_, _ = Ask(ctx, pid1, new(testpb.TestReply), askTimeout)
+			time.Sleep(time.Second)
+			wg.Done()
+		}()
+
+		// now we complete the Task
+		task <- new(testspb.TaskComplete)
+		wg.Wait()
+
+		require.EqualValues(t, 3, actor1.Counter())
+		require.EqualValues(t, 1, actor2.Counter())
+
+		time.Sleep(time.Second)
+		assert.NoError(t, pid1.Shutdown(ctx))
+		assert.NoError(t, pid2.Shutdown(ctx))
+	})
+	t.Run("With failed PipeTo", func(t *testing.T) {
+		askTimeout := time.Minute
+		ctx := context.TODO()
+
+		opts := []pidOption{
+			withInitMaxRetries(1),
+			withReplyTimeout(askTimeout),
+			withPassivationDisabled(),
+			withCustomLogger(log.DiscardLogger),
+		}
+
+		// create actor1
+		actor1 := &exchanger{}
+		actorPath1 := NewPath("Exchange1", NewAddress("sys", "host", 1))
+		pid1, err := newPID(ctx, actorPath1, actor1, opts...)
+		require.NoError(t, err)
+		require.NotNil(t, pid1)
+
+		// create actor2
+		actor2 := &exchanger{}
+		actorPath2 := NewPath("Exchange2", NewAddress("sys", "host", 1))
+		pid2, err := newPID(ctx, actorPath2, actor2, opts...)
+		require.NoError(t, err)
+		require.NotNil(t, pid2)
+
+		// zero message received by both actors
+		require.Zero(t, actor1.Counter())
+		require.Zero(t, actor2.Counter())
+
+		// create an instance of receive context
+		messageContext := &receiveContext{
+			ctx:            ctx,
+			message:        new(testpb.TaskComplete),
+			sender:         NoSender,
+			recipient:      pid1,
+			mu:             sync.Mutex{},
+			isAsyncMessage: true,
+		}
+
+		assert.Panics(t, func() {
+			messageContext.PipeTo(pid2, nil)
+		})
+
+		assert.NoError(t, pid1.Shutdown(ctx))
+		assert.NoError(t, pid2.Shutdown(ctx))
+	})
 }
diff --git a/actors/pid_option_test.go b/actors/pid_option_test.go
index c7c43f8..8396e1f 100644
--- a/actors/pid_option_test.go
+++ b/actors/pid_option_test.go
@@ -63,7 +63,7 @@ func TestPIDOptions(t *testing.T) {
 		},
 		{
 			name:     "WithSendReplyTimeout",
-			option:   withSendReplyTimeout(time.Second),
+			option:   withReplyTimeout(time.Second),
 			expected: &pid{replyTimeout: atomicDuration},
 		},
 		{
diff --git a/actors/pid_test.go b/actors/pid_test.go
index 94d8049..7e43ead 100644
--- a/actors/pid_test.go
+++ b/actors/pid_test.go
@@ -45,12 +45,13 @@ import (
 	"github.com/tochemey/goakt/v2/log"
 	"github.com/tochemey/goakt/v2/telemetry"
 	"github.com/tochemey/goakt/v2/test/data/testpb"
+	testspb "github.com/tochemey/goakt/v2/test/data/testpb"
 )
 
 const (
-	receivingDelay   = 1 * time.Second
-	receivingTimeout = 100 * time.Millisecond
-	passivateAfter   = 200 * time.Millisecond
+	receivingDelay = 1 * time.Second
+	replyTimeout   = 100 * time.Millisecond
+	passivateAfter = 200 * time.Millisecond
 )
 
 func TestReceive(t *testing.T) {
@@ -66,7 +67,7 @@ func TestReceive(t *testing.T) {
 		newTestActor(),
 		withInitMaxRetries(1),
 		withCustomLogger(log.DefaultLogger),
-		withSendReplyTimeout(receivingTimeout))
+		withReplyTimeout(replyTimeout))
 
 	require.NoError(t, err)
 	assert.NotNil(t, pid)
@@ -96,7 +97,7 @@ func TestPassivation(t *testing.T) {
 		opts := []pidOption{
 			withInitMaxRetries(1),
 			withPassivationAfter(passivateAfter),
-			withSendReplyTimeout(receivingTimeout),
+			withReplyTimeout(replyTimeout),
 		}
 
 		// create the actor path
@@ -125,7 +126,7 @@ func TestPassivation(t *testing.T) {
 		opts := []pidOption{
 			withInitMaxRetries(1),
 			withPassivationAfter(passivateAfter),
-			withSendReplyTimeout(receivingTimeout),
+			withReplyTimeout(replyTimeout),
 		}
 
 		// create the actor path
@@ -165,7 +166,7 @@ func TestReply(t *testing.T) {
 		require.NoError(t, err)
 		assert.NotNil(t, pid)
 
-		actual, err := Ask(ctx, pid, new(testpb.TestReply), receivingTimeout)
+		actual, err := Ask(ctx, pid, new(testpb.TestReply), replyTimeout)
 		assert.NoError(t, err)
 		assert.NotNil(t, actual)
 		expected := &testpb.Reply{Content: "received message"}
@@ -189,7 +190,7 @@ func TestReply(t *testing.T) {
 		require.NoError(t, err)
 		assert.NotNil(t, pid)
 
-		actual, err := Ask(ctx, pid, new(testpb.TestSend), receivingTimeout)
+		actual, err := Ask(ctx, pid, new(testpb.TestSend), replyTimeout)
 		assert.Error(t, err)
 		assert.EqualError(t, err, ErrRequestTimeout.Error())
 		assert.Nil(t, actual)
@@ -216,7 +217,7 @@ func TestReply(t *testing.T) {
 		err = pid.Shutdown(ctx)
 		assert.NoError(t, err)
 
-		actual, err := Ask(ctx, pid, new(testpb.TestReply), receivingTimeout)
+		actual, err := Ask(ctx, pid, new(testpb.TestReply), replyTimeout)
 		assert.Error(t, err)
 		assert.EqualError(t, err, ErrDead.Error())
 		assert.Nil(t, actual)
@@ -237,7 +238,7 @@ func TestRestart(t *testing.T) {
 			withInitMaxRetries(1),
 			withPassivationAfter(10*time.Second),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, pid)
@@ -281,7 +282,7 @@ func TestRestart(t *testing.T) {
 		pid, err := newPID(ctx, actorPath, actor,
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, pid)
@@ -320,7 +321,7 @@ func TestRestart(t *testing.T) {
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
 			withPassivationAfter(time.Minute),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, pid)
@@ -360,7 +361,7 @@ func TestRestart(t *testing.T) {
 			withInitMaxRetries(1),
 			withPassivationAfter(passivateAfter),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, pid)
@@ -389,7 +390,7 @@ func TestSupervisorStrategy(t *testing.T) {
 			newSupervisor(),
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -424,7 +425,7 @@ func TestSupervisorStrategy(t *testing.T) {
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
 			withPassivationDisabled(),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -465,7 +466,7 @@ func TestSupervisorStrategy(t *testing.T) {
 			withCustomLogger(log.DiscardLogger),
 			withPassivationDisabled(),
 			withSupervisorStrategy(-1), // this is a rogue strategy which will default to a Stop
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -506,7 +507,7 @@ func TestSupervisorStrategy(t *testing.T) {
 			withCustomLogger(log.DiscardLogger),
 			withSupervisorStrategy(DefaultSupervisoryStrategy),
 			withPassivationDisabled(),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -548,7 +549,7 @@ func TestSupervisorStrategy(t *testing.T) {
 			withCustomLogger(logger),
 			withPassivationDisabled(),
 			withSupervisorStrategy(RestartDirective),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -586,7 +587,7 @@ func TestSupervisorStrategy(t *testing.T) {
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
 			withPassivationDisabled(),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -742,7 +743,7 @@ func TestMessaging(t *testing.T) {
 		opts := []pidOption{
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout),
+			withReplyTimeout(replyTimeout),
 		}
 
 		// create the actor path
@@ -860,7 +861,7 @@ func TestActorHandle(t *testing.T) {
 		&exchanger{},
 		withInitMaxRetries(1),
 		withCustomLogger(log.DefaultLogger),
-		withSendReplyTimeout(receivingTimeout))
+		withReplyTimeout(replyTimeout))
 
 	require.NoError(t, err)
 	assert.NotNil(t, pid)
@@ -885,7 +886,7 @@ func TestPIDActorSystem(t *testing.T) {
 		&exchanger{},
 		withInitMaxRetries(1),
 		withCustomLogger(log.DefaultLogger),
-		withSendReplyTimeout(receivingTimeout))
+		withReplyTimeout(replyTimeout))
 	require.NoError(t, err)
 	assert.NotNil(t, pid)
 	sys := pid.ActorSystem()
@@ -906,7 +907,7 @@ func TestSpawnChild(t *testing.T) {
 			newSupervisor(),
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -945,7 +946,7 @@ func TestSpawnChild(t *testing.T) {
 			newSupervisor(),
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -981,7 +982,7 @@ func TestSpawnChild(t *testing.T) {
 			newSupervisor(),
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -1008,7 +1009,7 @@ func TestSpawnChild(t *testing.T) {
 			newSupervisor(),
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -1037,7 +1038,7 @@ func TestPoisonPill(t *testing.T) {
 		newTestActor(),
 		withInitMaxRetries(1),
 		withCustomLogger(log.DefaultLogger),
-		withSendReplyTimeout(receivingTimeout))
+		withReplyTimeout(replyTimeout))
 
 	require.NoError(t, err)
 	assert.NotNil(t, pid)
@@ -1173,7 +1174,7 @@ func TestFailedPostStop(t *testing.T) {
 		&testPostStop{},
 		withInitMaxRetries(1),
 		withCustomLogger(log.DefaultLogger),
-		withSendReplyTimeout(receivingTimeout))
+		withReplyTimeout(replyTimeout))
 
 	require.NoError(t, err)
 	assert.NotNil(t, pid)
@@ -1192,7 +1193,7 @@ func TestShutdown(t *testing.T) {
 			newSupervisor(),
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, parent)
@@ -1343,7 +1344,7 @@ func TestBatchAsk(t *testing.T) {
 		opts := []pidOption{
 			withInitMaxRetries(1),
 			withCustomLogger(log.DiscardLogger),
-			withSendReplyTimeout(receivingTimeout),
+			withReplyTimeout(replyTimeout),
 		}
 
 		// create the actor path
@@ -1383,7 +1384,7 @@ func TestRegisterMetrics(t *testing.T) {
 		withCustomLogger(log.DefaultLogger),
 		withTelemetry(tel),
 		withMetric(),
-		withSendReplyTimeout(receivingTimeout))
+		withReplyTimeout(replyTimeout))
 
 	require.NoError(t, err)
 	assert.NotNil(t, pid)
@@ -1508,7 +1509,6 @@ func TestRemoteReSpawn(t *testing.T) {
 		})
 	})
 }
-
 func TestRemoteStop(t *testing.T) {
 	t.Run("With actor address not found", func(t *testing.T) {
 		// create the context
@@ -1629,7 +1629,6 @@ func TestRemoteStop(t *testing.T) {
 		})
 	})
 }
-
 func TestID(t *testing.T) {
 	ctx := context.TODO()
 	// create the actor path
@@ -1642,7 +1641,7 @@ func TestID(t *testing.T) {
 		&exchanger{},
 		withInitMaxRetries(1),
 		withCustomLogger(log.DiscardLogger),
-		withSendReplyTimeout(receivingTimeout))
+		withReplyTimeout(replyTimeout))
 	require.NoError(t, err)
 	assert.NotNil(t, pid)
 	sys := pid.ActorSystem()
@@ -1657,7 +1656,6 @@ func TestID(t *testing.T) {
 	err = pid.Shutdown(ctx)
 	assert.NoError(t, err)
 }
-
 func TestEquals(t *testing.T) {
 	ctx := context.TODO()
 	logger := log.DiscardLogger
@@ -1682,7 +1680,6 @@ func TestEquals(t *testing.T) {
 	err = sys.Stop(ctx)
 	assert.NoError(t, err)
 }
-
 func TestRemoteSpawn(t *testing.T) {
 	t.Run("When remoting is enabled", func(t *testing.T) {
 		// create the context
@@ -1842,7 +1839,6 @@ func TestRemoteSpawn(t *testing.T) {
 		})
 	})
 }
-
 func TestName(t *testing.T) {
 	ctx := context.TODO()
 	// create the actor path
@@ -1855,7 +1851,7 @@ func TestName(t *testing.T) {
 		&exchanger{},
 		withInitMaxRetries(1),
 		withCustomLogger(log.DiscardLogger),
-		withSendReplyTimeout(receivingTimeout))
+		withReplyTimeout(replyTimeout))
 	require.NoError(t, err)
 	assert.NotNil(t, pid)
 	sys := pid.ActorSystem()
@@ -1870,3 +1866,251 @@ func TestName(t *testing.T) {
 	err = pid.Shutdown(ctx)
 	assert.NoError(t, err)
 }
+func TestPipeTo(t *testing.T) {
+	t.Run("With happy path", func(t *testing.T) {
+		askTimeout := time.Minute
+		ctx := context.TODO()
+
+		opts := []pidOption{
+			withInitMaxRetries(1),
+			withReplyTimeout(askTimeout),
+			withPassivationDisabled(),
+			withCustomLogger(log.DefaultLogger),
+		}
+
+		// create actor1
+		actor1 := &exchanger{}
+		actorPath1 := NewPath("Exchange1", NewAddress("sys", "host", 1))
+		pid1, err := newPID(ctx, actorPath1, actor1, opts...)
+		require.NoError(t, err)
+		require.NotNil(t, pid1)
+
+		// create actor2
+		actor2 := &exchanger{}
+		actorPath2 := NewPath("Exchange2", NewAddress("sys", "host", 1))
+		pid2, err := newPID(ctx, actorPath2, actor2, opts...)
+		require.NoError(t, err)
+		require.NotNil(t, pid2)
+
+		// zero message received by both actors
+		require.Zero(t, actor1.Counter())
+		require.Zero(t, actor2.Counter())
+
+		task := make(chan proto.Message)
+		err = pid1.PipeTo(ctx, pid2, task)
+		require.NoError(t, err)
+
+		var wg sync.WaitGroup
+		wg.Add(1)
+		go func() {
+			// Wait for some time and during that period send some messages to the actor
+			// send three messages while waiting for the future to completed
+			_, _ = Ask(ctx, pid1, new(testpb.TestReply), askTimeout)
+			_, _ = Ask(ctx, pid1, new(testpb.TestReply), askTimeout)
+			_, _ = Ask(ctx, pid1, new(testpb.TestReply), askTimeout)
+			time.Sleep(time.Second)
+			wg.Done()
+		}()
+
+		// now we complete the Task
+		task <- new(testspb.TaskComplete)
+		wg.Wait()
+
+		require.EqualValues(t, 3, actor1.Counter())
+		require.EqualValues(t, 1, actor2.Counter())
+
+		time.Sleep(time.Second)
+		assert.NoError(t, pid1.Shutdown(ctx))
+		assert.NoError(t, pid2.Shutdown(ctx))
+	})
+	t.Run("With is a dead actor: case 1", func(t *testing.T) {
+		askTimeout := time.Minute
+		ctx := context.TODO()
+
+		opts := []pidOption{
+			withInitMaxRetries(1),
+			withReplyTimeout(askTimeout),
+			withPassivationDisabled(),
+			withCustomLogger(log.DiscardLogger),
+		}
+
+		// create actor1
+		actor1 := &exchanger{}
+		actorPath1 := NewPath("Exchange1", NewAddress("sys", "host", 1))
+		pid1, err := newPID(ctx, actorPath1, actor1, opts...)
+		require.NoError(t, err)
+		require.NotNil(t, pid1)
+
+		// create actor2
+		actor2 := &exchanger{}
+		actorPath2 := NewPath("Exchange2", NewAddress("sys", "host", 1))
+		pid2, err := newPID(ctx, actorPath2, actor2, opts...)
+		require.NoError(t, err)
+		require.NotNil(t, pid2)
+
+		// shutdown the actor after one second of liveness
+		time.Sleep(time.Second)
+		assert.NoError(t, pid2.Shutdown(ctx))
+
+		task := make(chan proto.Message)
+		err = pid1.PipeTo(ctx, pid2, task)
+		require.Error(t, err)
+		assert.EqualError(t, err, ErrDead.Error())
+
+		time.Sleep(time.Second)
+		assert.NoError(t, pid1.Shutdown(ctx))
+	})
+	t.Run("With is a dead actor: case 2", func(t *testing.T) {
+		askTimeout := time.Minute
+		ctx := context.TODO()
+
+		opts := []pidOption{
+			withInitMaxRetries(1),
+			withReplyTimeout(askTimeout),
+			withPassivationDisabled(),
+			withCustomLogger(log.DiscardLogger),
+		}
+
+		// create actor1
+		actor1 := &exchanger{}
+		actorPath1 := NewPath("Exchange1", NewAddress("sys", "host", 1))
+		pid1, err := newPID(ctx, actorPath1, actor1, opts...)
+		require.NoError(t, err)
+		require.NotNil(t, pid1)
+
+		// create actor2
+		actor2 := &exchanger{}
+		actorPath2 := NewPath("Exchange2", NewAddress("sys", "host", 1))
+		pid2, err := newPID(ctx, actorPath2, actor2, opts...)
+		require.NoError(t, err)
+		require.NotNil(t, pid2)
+
+		// zero message received by both actors
+		require.Zero(t, actor1.Counter())
+		require.Zero(t, actor2.Counter())
+
+		task := make(chan proto.Message)
+		err = pid1.PipeTo(ctx, pid2, task)
+		require.NoError(t, err)
+
+		var wg sync.WaitGroup
+		wg.Add(1)
+		go func() {
+			// Wait for some time and during that period send some messages to the actor
+			// send three messages while waiting for the future to completed
+			_, _ = Ask(ctx, pid1, new(testpb.TestReply), askTimeout)
+			_, _ = Ask(ctx, pid1, new(testpb.TestReply), askTimeout)
+			_, _ = Ask(ctx, pid1, new(testpb.TestReply), askTimeout)
+			time.Sleep(time.Second)
+			wg.Done()
+		}()
+
+		// now we complete the Task
+		task <- new(testspb.TaskComplete)
+		_ = Tell(ctx, pid2, new(testpb.TestBye))
+
+		wg.Wait()
+
+		require.EqualValues(t, 3, actor1.Counter())
+		require.Zero(t, actor2.Counter())
+
+		time.Sleep(time.Second)
+		assert.NoError(t, pid1.Shutdown(ctx))
+	})
+	t.Run("With undefined task", func(t *testing.T) {
+		askTimeout := time.Minute
+		ctx := context.TODO()
+
+		opts := []pidOption{
+			withInitMaxRetries(1),
+			withReplyTimeout(askTimeout),
+			withPassivationDisabled(),
+			withCustomLogger(log.DiscardLogger),
+		}
+
+		// create actor1
+		actor1 := &exchanger{}
+		actorPath1 := NewPath("Exchange1", NewAddress("sys", "host", 1))
+		pid1, err := newPID(ctx, actorPath1, actor1, opts...)
+		require.NoError(t, err)
+		require.NotNil(t, pid1)
+
+		// create actor2
+		actor2 := &exchanger{}
+		actorPath2 := NewPath("Exchange2", NewAddress("sys", "host", 1))
+		pid2, err := newPID(ctx, actorPath2, actor2, opts...)
+		require.NoError(t, err)
+		require.NotNil(t, pid2)
+
+		// zero message received by both actors
+		require.Zero(t, actor1.Counter())
+		require.Zero(t, actor2.Counter())
+
+		err = pid1.PipeTo(ctx, pid2, nil)
+		require.Error(t, err)
+		assert.EqualError(t, err, ErrUndefinedTask.Error())
+
+		time.Sleep(time.Second)
+		assert.NoError(t, pid1.Shutdown(ctx))
+		assert.NoError(t, pid2.Shutdown(ctx))
+	})
+	t.Run("With failed task", func(t *testing.T) {
+		askTimeout := time.Minute
+		ctx := context.TODO()
+
+		opts := []pidOption{
+			withInitMaxRetries(1),
+			withReplyTimeout(askTimeout),
+			withPassivationDisabled(),
+			withCustomLogger(log.DiscardLogger),
+		}
+
+		// create actor1
+		actor1 := &exchanger{}
+		actorPath1 := NewPath("Exchange1", NewAddress("sys", "host", 1))
+		pid1, err := newPID(ctx, actorPath1, actor1, opts...)
+		require.NoError(t, err)
+		require.NotNil(t, pid1)
+
+		// create actor2
+		actor2 := &exchanger{}
+		actorPath2 := NewPath("Exchange2", NewAddress("sys", "host", 1))
+		pid2, err := newPID(ctx, actorPath2, actor2, opts...)
+		require.NoError(t, err)
+		require.NotNil(t, pid2)
+
+		// zero message received by both actors
+		require.Zero(t, actor1.Counter())
+		require.Zero(t, actor2.Counter())
+
+		task := make(chan proto.Message)
+
+		cancelCtx, cancel := context.WithCancel(ctx)
+		err = pid1.PipeTo(cancelCtx, pid2, task)
+		require.NoError(t, err)
+
+		var wg sync.WaitGroup
+		wg.Add(1)
+		go func() {
+			// Wait for some time and during that period send some messages to the actor
+			// send three messages while waiting for the future to completed
+			_, _ = Ask(ctx, pid1, new(testpb.TestReply), askTimeout)
+			_, _ = Ask(ctx, pid1, new(testpb.TestReply), askTimeout)
+			_, _ = Ask(ctx, pid1, new(testpb.TestReply), askTimeout)
+			time.Sleep(time.Second)
+			wg.Done()
+		}()
+
+		cancel()
+		wg.Wait()
+
+		require.EqualValues(t, 3, actor1.Counter())
+
+		// no message piped to the actor
+		require.Zero(t, actor2.Counter())
+
+		time.Sleep(time.Second)
+		assert.NoError(t, pid1.Shutdown(ctx))
+		assert.NoError(t, pid2.Shutdown(ctx))
+	})
+}
diff --git a/actors/scheduler_test.go b/actors/scheduler_test.go
index 3d3e9b5..b3eb413 100644
--- a/actors/scheduler_test.go
+++ b/actors/scheduler_test.go
@@ -96,7 +96,7 @@ func TestScheduler(t *testing.T) {
 			newTestActor(),
 			withInitMaxRetries(1),
 			withCustomLogger(log.DefaultLogger),
-			withSendReplyTimeout(receivingTimeout))
+			withReplyTimeout(replyTimeout))
 
 		require.NoError(t, err)
 		assert.NotNil(t, pid)
diff --git a/future/future_test.go b/future/future_test.go
new file mode 100644
index 0000000..906b2d3
--- /dev/null
+++ b/future/future_test.go
@@ -0,0 +1,170 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2022-2024 Tochemey
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package future
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/tochemey/goakt/v2/test/data/testpb"
+)
+
+func TestMain(m *testing.M) {
+	goleak.VerifyTestMain(m)
+}
+
+func TestWaitOnResult(t *testing.T) {
+	executor := make(chan proto.Message)
+	f := New(executor, time.Duration(30*time.Minute))
+	var result *Result
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		result = f.Result()
+		wg.Done()
+	}()
+
+	executor <- new(testpb.Ping)
+	wg.Wait()
+
+	assert.Nil(t, result.Failure())
+	assert.True(t, proto.Equal(new(testpb.Ping), result.Success()))
+
+	// ensure we don't get paused on the next iteration.
+	result = f.Result()
+	assert.True(t, proto.Equal(new(testpb.Ping), result.Success()))
+	assert.Nil(t, result.Failure())
+}
+
+func TestHasResult(t *testing.T) {
+	t.Run("With timeout set", func(t *testing.T) {
+		executor := make(chan proto.Message)
+		f := New(executor, time.Duration(30*time.Minute))
+
+		assert.False(t, f.HasResult())
+
+		var wg sync.WaitGroup
+		wg.Add(1)
+		go func() {
+			f.Result()
+			wg.Done()
+		}()
+
+		executor <- new(testpb.Ping)
+		wg.Wait()
+
+		assert.True(t, f.HasResult())
+	})
+	t.Run("With context set", func(t *testing.T) {
+		ctx := context.TODO()
+		executor := make(chan proto.Message)
+		f := NewWithContext(ctx, executor)
+
+		assert.False(t, f.HasResult())
+
+		var wg sync.WaitGroup
+		wg.Add(1)
+		go func() {
+			f.Result()
+			wg.Done()
+		}()
+
+		executor <- new(testpb.Ping)
+		wg.Wait()
+
+		assert.True(t, f.HasResult())
+	})
+}
+
+func TestTimeout(t *testing.T) {
+	executor := make(chan proto.Message)
+	f := New(executor, time.Duration(0))
+
+	result := f.Result()
+
+	require.NotNil(t, result)
+	assert.Nil(t, result.Success())
+	assert.NotNil(t, result.Failure())
+}
+
+func TestContextCancelation(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.TODO())
+	executor := make(chan proto.Message)
+	f := NewWithContext(ctx, executor)
+	cancel()
+
+	result := f.Result()
+
+	require.NotNil(t, result)
+	assert.Nil(t, result.Success())
+	assert.NotNil(t, result.Failure())
+}
+
+func BenchmarkFuture(b *testing.B) {
+	executor := make(chan proto.Message)
+	timeout := time.Duration(30 * time.Minute)
+	var wg sync.WaitGroup
+
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		wg.Add(1)
+		f := New(executor, timeout)
+		go func() {
+			f.Result()
+			wg.Done()
+		}()
+
+		executor <- new(testpb.Ping)
+		wg.Wait()
+	}
+}
+
+func BenchmarkFutureWithContext(b *testing.B) {
+	executor := make(chan proto.Message)
+	ctx := context.TODO()
+	var wg sync.WaitGroup
+
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		wg.Add(1)
+		f := NewWithContext(ctx, executor)
+		go func() {
+			f.Result()
+			wg.Done()
+		}()
+
+		executor <- new(testpb.Ping)
+		wg.Wait()
+	}
+}
diff --git a/protos/test/test.proto b/protos/test/test.proto
index bcb9e1a..2d4cb9d 100644
--- a/protos/test/test.proto
+++ b/protos/test/test.proto
@@ -57,3 +57,8 @@ message Pong {}
 message Wait {
   uint64 duration = 1;
 }
+
+message RunTask {
+  uint64 duration = 1;
+}
+message TaskComplete {}
diff --git a/test/data/testpb/test.pb.go b/test/data/testpb/test.pb.go
index de4c167..995348d 100644
--- a/test/data/testpb/test.pb.go
+++ b/test/data/testpb/test.pb.go
@@ -1027,6 +1027,91 @@ func (x *Wait) GetDuration() uint64 {
 	return 0
 }
 
+type RunTask struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Duration uint64 `protobuf:"varint,1,opt,name=duration,proto3" json:"duration,omitempty"`
+}
+
+func (x *RunTask) Reset() {
+	*x = RunTask{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_test_test_proto_msgTypes[24]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *RunTask) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*RunTask) ProtoMessage() {}
+
+func (x *RunTask) ProtoReflect() protoreflect.Message {
+	mi := &file_test_test_proto_msgTypes[24]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use RunTask.ProtoReflect.Descriptor instead.
+func (*RunTask) Descriptor() ([]byte, []int) {
+	return file_test_test_proto_rawDescGZIP(), []int{24}
+}
+
+func (x *RunTask) GetDuration() uint64 {
+	if x != nil {
+		return x.Duration
+	}
+	return 0
+}
+
+type TaskComplete struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+}
+
+func (x *TaskComplete) Reset() {
+	*x = TaskComplete{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_test_test_proto_msgTypes[25]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *TaskComplete) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*TaskComplete) ProtoMessage() {}
+
+func (x *TaskComplete) ProtoReflect() protoreflect.Message {
+	mi := &file_test_test_proto_msgTypes[25]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use TaskComplete.ProtoReflect.Descriptor instead.
+func (*TaskComplete) Descriptor() ([]byte, []int) {
+	return file_test_test_proto_rawDescGZIP(), []int{25}
+}
+
 var File_test_test_proto protoreflect.FileDescriptor
 
 var file_test_test_proto_rawDesc = []byte{
@@ -1075,16 +1160,19 @@ var file_test_test_proto_rawDesc = []byte{
 	0x6e, 0x73, 0x74, 0x61, 0x73, 0x68, 0x41, 0x6c, 0x6c, 0x22, 0x06, 0x0a, 0x04, 0x50, 0x69, 0x6e,
 	0x67, 0x22, 0x06, 0x0a, 0x04, 0x50, 0x6f, 0x6e, 0x67, 0x22, 0x22, 0x0a, 0x04, 0x57, 0x61, 0x69,
 	0x74, 0x12, 0x1a, 0x0a, 0x08, 0x64, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20,
-	0x01, 0x28, 0x04, 0x52, 0x08, 0x64, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x42, 0x80, 0x01,
-	0x0a, 0x0a, 0x63, 0x6f, 0x6d, 0x2e, 0x74, 0x65, 0x73, 0x74, 0x70, 0x62, 0x42, 0x09, 0x54, 0x65,
-	0x73, 0x74, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x48, 0x02, 0x50, 0x01, 0x5a, 0x2d, 0x67, 0x69, 0x74,
-	0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x74, 0x6f, 0x63, 0x68, 0x65, 0x6d, 0x65, 0x79,
-	0x2f, 0x67, 0x6f, 0x61, 0x6b, 0x74, 0x2f, 0x76, 0x32, 0x2f, 0x74, 0x65, 0x73, 0x74, 0x2f, 0x64,
-	0x61, 0x74, 0x61, 0x3b, 0x74, 0x65, 0x73, 0x74, 0x70, 0x62, 0xa2, 0x02, 0x03, 0x54, 0x58, 0x58,
-	0xaa, 0x02, 0x06, 0x54, 0x65, 0x73, 0x74, 0x70, 0x62, 0xca, 0x02, 0x06, 0x54, 0x65, 0x73, 0x74,
-	0x70, 0x62, 0xe2, 0x02, 0x12, 0x54, 0x65, 0x73, 0x74, 0x70, 0x62, 0x5c, 0x47, 0x50, 0x42, 0x4d,
-	0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0xea, 0x02, 0x06, 0x54, 0x65, 0x73, 0x74, 0x70, 0x62,
-	0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x01, 0x28, 0x04, 0x52, 0x08, 0x64, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0x25, 0x0a,
+	0x07, 0x52, 0x75, 0x6e, 0x54, 0x61, 0x73, 0x6b, 0x12, 0x1a, 0x0a, 0x08, 0x64, 0x75, 0x72, 0x61,
+	0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x04, 0x52, 0x08, 0x64, 0x75, 0x72, 0x61,
+	0x74, 0x69, 0x6f, 0x6e, 0x22, 0x0e, 0x0a, 0x0c, 0x54, 0x61, 0x73, 0x6b, 0x43, 0x6f, 0x6d, 0x70,
+	0x6c, 0x65, 0x74, 0x65, 0x42, 0x80, 0x01, 0x0a, 0x0a, 0x63, 0x6f, 0x6d, 0x2e, 0x74, 0x65, 0x73,
+	0x74, 0x70, 0x62, 0x42, 0x09, 0x54, 0x65, 0x73, 0x74, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x48, 0x02,
+	0x50, 0x01, 0x5a, 0x2d, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x74,
+	0x6f, 0x63, 0x68, 0x65, 0x6d, 0x65, 0x79, 0x2f, 0x67, 0x6f, 0x61, 0x6b, 0x74, 0x2f, 0x76, 0x32,
+	0x2f, 0x74, 0x65, 0x73, 0x74, 0x2f, 0x64, 0x61, 0x74, 0x61, 0x3b, 0x74, 0x65, 0x73, 0x74, 0x70,
+	0x62, 0xa2, 0x02, 0x03, 0x54, 0x58, 0x58, 0xaa, 0x02, 0x06, 0x54, 0x65, 0x73, 0x74, 0x70, 0x62,
+	0xca, 0x02, 0x06, 0x54, 0x65, 0x73, 0x74, 0x70, 0x62, 0xe2, 0x02, 0x12, 0x54, 0x65, 0x73, 0x74,
+	0x70, 0x62, 0x5c, 0x47, 0x50, 0x42, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0xea, 0x02,
+	0x06, 0x54, 0x65, 0x73, 0x74, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -1099,7 +1187,7 @@ func file_test_test_proto_rawDescGZIP() []byte {
 	return file_test_test_proto_rawDescData
 }
 
-var file_test_test_proto_msgTypes = make([]protoimpl.MessageInfo, 24)
+var file_test_test_proto_msgTypes = make([]protoimpl.MessageInfo, 26)
 var file_test_test_proto_goTypes = []interface{}{
 	(*TestReply)(nil),        // 0: testpb.TestReply
 	(*TestPanic)(nil),        // 1: testpb.TestPanic
@@ -1125,6 +1213,8 @@ var file_test_test_proto_goTypes = []interface{}{
 	(*Ping)(nil),             // 21: testpb.Ping
 	(*Pong)(nil),             // 22: testpb.Pong
 	(*Wait)(nil),             // 23: testpb.Wait
+	(*RunTask)(nil),          // 24: testpb.RunTask
+	(*TaskComplete)(nil),     // 25: testpb.TaskComplete
 }
 var file_test_test_proto_depIdxs = []int32{
 	0, // [0:0] is the sub-list for method output_type
@@ -1428,6 +1518,30 @@ func file_test_test_proto_init() {
 				return nil
 			}
 		}
+		file_test_test_proto_msgTypes[24].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*RunTask); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_test_test_proto_msgTypes[25].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*TaskComplete); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
 	}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
@@ -1435,7 +1549,7 @@ func file_test_test_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_test_test_proto_rawDesc,
 			NumEnums:      0,
-			NumMessages:   24,
+			NumMessages:   26,
 			NumExtensions: 0,
 			NumServices:   0,
 		},