PipeTo

PipeTo runs a task asynchronously and delivers its result (or failure) to a target actor’s mailbox. Use it to offload I/O or CPU-bound work without blocking the actor, while keeping the result in the actor model. Note: PipeTo can only be invoked by local actors. Calling pid.PipeTo or ctx.PipeTo when the sender (pid or ctx.Self()) is a remote PID returns ErrNotLocal. The task runs in a goroutine on the caller’s node; remote PIDs do not have a local execution context to run it.

API

From ReceiveContext

ctx.PipeTo: Run task in a goroutine; on success, send the result to to. On failure, behavior depends on options.
ctx.PipeToName: Same, but target is resolved by name (location-transparent).

From PID

pid.PipeTo: Run task; deliver result to to. The sender (pid) is used for context.
pid.PipeToName: Same, target by name.

From GrainContext

gctx.PipeToGrain: Deliver result to a grain.
gctx.PipeToActor: Deliver result to a named actor.
gctx.PipeToSelf: Deliver result to this grain.

Failure semantics differ for grain targets: a failed task delivers a *StatusFailure message to the target grain instead of going to the dead-letter queue. PipeToActor follows the actor semantics below. See Grain PipeTo for the grain-side API and example.

Task signature

task func() (any, error)

On success: the any result is sent to the target as a normal message.
On failure: the error is forwarded to the dead-letter queue, unless WithCircuitBreaker is used and the breaker is open (in which case the outcome is dropped).

The task runs on its own goroutine, not on the mailbox thread, so it must not read or write actor state. A field the task reads while a later message handler writes it is a genuine data race, and it is the one way to reintroduce the problem the actor model removes. Copy what the task needs into local variables first, return the result, and mutate state when the result arrives as a message.

Task signature

// Wrong: the closure reads actor state from another goroutine.
ctx.PipeTo(ctx.Self(), func() (any, error) {
    return http.Get(a.endpoint)
})

// Right: capture by value, then run.
endpoint := a.endpoint

ctx.PipeTo(ctx.Self(), func() (any, error) {
    return http.Get(endpoint)
})

PipeOption

WithTimeout: Abort delivery if the task does not complete within d.
WithCircuitBreaker: If the circuit breaker is open, do not deliver; drop the outcome.

Only one of WithTimeout or WithCircuitBreaker may be used per call. Using both returns ErrOnlyOneOptionAllowed.

Example

func (a *FetcherActor) Receive(ctx *ReceiveContext) {
    switch msg := ctx.Message().(type) {
    case *FetchURL:
        ctx.PipeToName("aggregator", func() (any, error) {
            return http.Get(msg.URL)
        }, actor.WithTimeout(5*time.Second))
    default:
        ctx.Unhandled()
    }
}

The aggregator actor receives the *http.Response (or the task fails and the error is handled per options).

Quickstart

Actors

Grains

AI Agents

Remoting

Clustering

Advanced

Architecture

API

From ReceiveContext

From PID

From GrainContext

Task signature

PipeOption

Example

​API

​From ReceiveContext

​From PID

​From GrainContext

​Task signature

​PipeOption

​Example

API

From ReceiveContext

From PID

From GrainContext

Task signature

PipeOption

Example