activity

Documentation

Overview

Package activity declares operations a service implements ("умею"): the targets of bindings and rules. Handle declares one run as a Temporal activity through the installed driver; workflows.Activity declares a child workflow through the native extension. Both are named as declared on the service's task queue, input and output enveloped in backplane.v1.ActivityCall / ActivityResult.

Inputs and outputs evolve additively: a reader ignores fields it does not know (protojson with DiscardUnknown), a missing field reads as its zero value, and renaming a field means adding a new one and keeping the old until every reader uses the new.

activity.Handle(root, "Send", smtp.Send, activity.StartToClose(10*time.Second),
    activity.Retry(activity.RetryHint{Attempts: 5}), activity.Describe("sends a mail"))

Index

• func Handle[Req, Res any](scope deps.Scope, name string, ...)
• func HandleSchema(scope deps.Scope, name string, in, out *sp.Schema, ...)
• func Heartbeat(ctx context.Context, details ...any)
• func IsNonRetryable(err error) bool
• func NonRetryable(err error) error
• func RetryAfter(err error, d time.Duration) error
• func RetryAfterOf(err error) (time.Duration, bool)
• type Info
  • func InfoOf(ctx context.Context) (Info, bool)
• type Option
  • func Describe(s string) Option
  • func HeartbeatTimeout(d time.Duration) Option
  • func Retry(h RetryHint) Option
  • func StartToClose(d time.Duration) Option
• type RetryHint

Functions

func Handle

func Handle[Req, Res any](
	scope deps.Scope, name string, fn func(ctx context.Context, in Req) (Res, error), opts ...Option,
)

Handle declares an activity and its implementation. name is CamelCase ([A-Z][A-Za-z0-9]*), anything else panics. fn's ctx carries InfoOf.

func HandleSchema

func HandleSchema(
	scope deps.Scope, name string, in, out *sp.Schema, fn func(ctx context.Context, in []byte) ([]byte, error),
	opts ...Option,
)

HandleSchema declares an activity whose input and output are known only at run time: in and out are their schemas (nil: plain JSON), fn gets the input as JSON bytes and answers JSON bytes. For a service that relays a contract it learns at startup (a plugin host, a gateway); a contract known at compile time uses Handle. fn's ctx carries InfoOf.

func Heartbeat

func Heartbeat(ctx context.Context, details ...any)

Heartbeat reports progress of a long handler: the step's heartbeat timeout (HeartbeatTimeout) counts from the last one, and a cancelled step is noticed by ctx on the next one. details show as the pending activity's heartbeat details in Temporal. Outside a transport run it does nothing.

func IsNonRetryable

func IsNonRetryable(err error) bool

IsNonRetryable reports whether err was marked final with NonRetryable.

func NonRetryable

func NonRetryable(err error) error

NonRetryable marks err as final: the binding does not retry the activity (on Temporal, a non-retryable application error). Any other error is retried by the binding's retry policy; a *temporal.ApplicationError is passed as is, so its own retryability applies. nil stays nil.

func RetryAfter

func RetryAfter(err error, d time.Duration) error

RetryAfter marks err retryable no sooner than d: the remote side asked to wait (a rate limit, a Retry-After header). The binding's next attempt waits at least d, more when its backoff is longer; a NonRetryable error stays final. nil stays nil; d <= 0 is a plain retryable error.

func RetryAfterOf

func RetryAfterOf(err error) (time.Duration, bool)

RetryAfterOf is the delay RetryAfter asked for; false when err carries none.

Types

type Info

type Info struct {
	// Attempt is 1 for the first execution, then counts retries.
	Attempt int32
	// Binding and Step name the binding or rule that called and its step,
	// as the caller put them in the envelope.
	Binding string
	Step    string
	// Key is the same for every attempt of one step execution and unique
	// across them: "<workflow id>/<activity id>". An idempotency key for
	// what the handler does outside.
	Key string
	// Deadline of this attempt; zero when unbounded.
	Deadline time.Time
}

Info is what a handler knows about the execution it serves.

func InfoOf

func InfoOf(ctx context.Context) (Info, bool)

InfoOf is the Info of the execution ctx belongs to; false outside a handler run by the transport (a test calling the handler directly).

type Option

type Option = decl.ActivityOption

Option sets what the manifest tells a binding about the activity: its defaults for a step calling it. The binding may override them; the handler does not enforce them.

func Describe

func Describe(s string) Option

Describe is the activity's description for the console.

func HeartbeatTimeout

func HeartbeatTimeout(d time.Duration) Option

HeartbeatTimeout: an attempt that does not Heartbeat within d is considered lost and retried. Only for activities that call Heartbeat.

func Retry

func Retry(h RetryHint) Option

Retry sets the default retry of a step calling the activity.

func StartToClose

func StartToClose(d time.Duration) Option

StartToClose is how long one attempt may take (for a workflow-backed activity, one run).

type RetryHint

type RetryHint struct {
	// Attempts in total, the first included; 0 is the platform's default.
	Attempts uint32
}

RetryHint is the retry a binding step should use by default.