workflow

Documentation

Index

• Variables
• func Go(ctx Context, f func(ctx Context))
• func Logger(ctx Context) log.Logger
• func Now(ctx sync.Context) time.Time
• func Replaying(ctx sync.Context) bool
• func Select(ctx Context, cases ...SelectCase)
• func Sleep(ctx sync.Context, d time.Duration) error
• func WithCancel(parent Context) (ctx Context, cancel CancelFunc)
• func WorkflowInstance(ctx sync.Context) *core.WorkflowInstance
• type ActivityOptions
• type CancelFunc
• type Channel
  • func NewBufferedChannel[T any](size int) Channel[T]
  • func NewChannel[T any]() Channel[T]
  • func NewSignalChannel[T any](ctx Context, name string) Channel[T]
• type Context
  • func NewDisconnectedContext(ctx Context) Context
• type Future
  • func CreateSubWorkflowInstance[TResult any](ctx sync.Context, options SubWorkflowOptions, workflow interface{}, ...) Future[TResult]
  • func ExecuteActivity[TResult any](ctx Context, options ActivityOptions, activity interface{}, ...) Future[TResult]
  • func ScheduleTimer(ctx Context, delay time.Duration) Future[struct{}]
  • func SideEffect[TResult any](ctx Context, f func(ctx Context) TResult) Future[TResult]
  • func SignalWorkflow[T any](ctx Context, instanceID string, name string, arg T) Future[any]
• type Instance
• type Metadata
• type RetryOptions
• type SelectCase
  • func Await[T any](f Future[T], handler func(Context, Future[T])) SelectCase
  • func Default(handler func(Context)) SelectCase
  • func Receive[T any](c Channel[T], handler func(ctx Context, v T, ok bool)) SelectCase
  • func Send[T any](c Channel[T], value *T, handler func(ctx Context)) SelectCase
• type Span
• type SubWorkflowOptions
• type WaitGroup
  • func NewWaitGroup() WaitGroup
• type Workflow
• type WorkflowTracer
  • func Tracer(ctx Context) WorkflowTracer

Variables

var (
	DefaultSubWorkflowRetryOptions = RetryOptions{

		MaxAttempts: 1,
	}

	DefaultSubWorkflowOptions = SubWorkflowOptions{
		RetryOptions: DefaultSubWorkflowRetryOptions,
	}
)

var Canceled = sync.Canceled

var DefaultActivityOptions = ActivityOptions{
	RetryOptions: DefaultRetryOptions,
}

var DefaultRetryOptions = RetryOptions{
	MaxAttempts:        3,
	BackoffCoefficient: 1,
}

Functions

func Go

func Go(ctx Context, f func(ctx Context))

Go spawns a workflow goroutine

func Logger

func Logger(ctx Context) log.Logger

func Now

func Now(ctx sync.Context) time.Time

func Replaying

func Replaying(ctx sync.Context) bool

func Select

func Select(ctx Context, cases ...SelectCase)

Select is the workflow-save equivalent of the select statement.

func Sleep

func Sleep(ctx sync.Context, d time.Duration) error

func WithCancel

func WithCancel(parent Context) (ctx Context, cancel CancelFunc)

WithCancel returns a copy of parent with a new Done channel. The returned context's Done channel is closed when the returned cancel function is called or when the parent context's Done channel is closed, whichever happens first.

Canceling this context releases resources associated with it, so code should call cancel as soon as the operations running in this Context complete.

func WorkflowInstance

func WorkflowInstance(ctx sync.Context) *core.WorkflowInstance

Types

type ActivityOptions

type ActivityOptions struct {
	RetryOptions RetryOptions
}

type CancelFunc

type CancelFunc = sync.CancelFunc

type Channel

type Channel[T any] interface {
	Send(ctx Context, v T)

	SendNonblocking(v T) (ok bool)

	Receive(ctx Context) (v T, ok bool)

	ReceiveNonBlocking() (v T, ok bool)

	Close()
}

func NewBufferedChannel

func NewBufferedChannel[T any](size int) Channel[T]

func NewChannel

func NewChannel[T any]() Channel[T]

func NewSignalChannel

func NewSignalChannel[T any](ctx Context, name string) Channel[T]

type Context

type Context = sync.Context

func NewDisconnectedContext

func NewDisconnectedContext(ctx Context) Context

type Future

type Future[T any] interface {
	// Get returns the value if set, blocks otherwise
	Get(ctx Context) (T, error)
}

func CreateSubWorkflowInstance

func CreateSubWorkflowInstance[TResult any](ctx sync.Context, options SubWorkflowOptions, workflow interface{}, args ...interface{}) Future[TResult]

func ExecuteActivity

func ExecuteActivity[TResult any](ctx Context, options ActivityOptions, activity interface{}, args ...interface{}) Future[TResult]

ExecuteActivity schedules the given activity to be executed

func ScheduleTimer

func ScheduleTimer(ctx Context, delay time.Duration) Future[struct{}]

func SideEffect

func SideEffect[TResult any](ctx Context, f func(ctx Context) TResult) Future[TResult]

func SignalWorkflow

func SignalWorkflow[T any](ctx Context, instanceID string, name string, arg T) Future[any]

type Instance

type Instance = core.WorkflowInstance

type Metadata

type Metadata = core.WorkflowMetadata

type RetryOptions

type RetryOptions struct {
	// Maximum number of times to retry
	MaxAttempts int

	// Time to wait before first retry
	FirstRetryInterval time.Duration

	// Maximum delay for any individual retry attempt
	MaxRetryInterval time.Duration

	// Coeffecient for calculation the next retry delay
	BackoffCoefficient float64

	// Timeout after which retries are aborted
	RetryTimeout time.Duration
}

type SelectCase

type SelectCase = sync.SelectCase

func Await

func Await[T any](f Future[T], handler func(Context, Future[T])) SelectCase

Await calls the provided handler when the given future is ready.

func Default

func Default(handler func(Context)) SelectCase

Default calls the given handler if none of the other cases match.

func Receive

func Receive[T any](c Channel[T], handler func(ctx Context, v T, ok bool)) SelectCase

Receive calls the provided handler if the given channel can receive a value. The handler receives the received value, and the ok flag indicating whether the value was received or the channel was closed.

func Send

func Send[T any](c Channel[T], value *T, handler func(ctx Context)) SelectCase

Send calls the provided handler if the given value can be sent to the channel.

type Span

type Span interface {
	End()
}

type SubWorkflowOptions

type SubWorkflowOptions struct {
	InstanceID string

	RetryOptions RetryOptions
}

type WaitGroup

type WaitGroup = sync.WaitGroup

func NewWaitGroup

func NewWaitGroup() WaitGroup

type Workflow

type Workflow = interface{}

type WorkflowTracer

type WorkflowTracer interface {
	Start(ctx Context, name string, opts ...trace.SpanStartOption) (Context, Span)
}

func Tracer

func Tracer(ctx Context) WorkflowTracer