Versions in this module Expand all Collapse all v0 v0.1.0 Sep 1, 2026 Changes in this version + func PortableValueAs[T any](v PortableValue) (T, bool) + type AddHandlerOption func(*addHandlerOptions) + func WithHandlerOverwrite(overwrite bool) AddHandlerOption + type AttrSendsMessage struct + type AttrYieldsOutput struct + type Builder struct + func NewBuilder(start ExecutorBinding) *Builder + func (wb *Builder) AddChain(source ExecutorBinding, executors []ExecutorBinding, allowRepetition bool) *Builder + func (wb *Builder) AddEdge(source ExecutorBinding, target ExecutorBinding, opts ...EdgeOption) *Builder + func (wb *Builder) AddFanInBarrierEdge(sources []ExecutorBinding, target ExecutorBinding, opts ...EdgeOption) *Builder + func (wb *Builder) AddFanOutEdge(source ExecutorBinding, targets []ExecutorBinding, opts ...EdgeOption) *Builder + func (wb *Builder) AddSwitch(source ExecutorBinding) *SwitchBuilder + func (wb *Builder) BindExecutor(binding ExecutorBinding) *Builder + func (wb *Builder) Build() (*Workflow, error) + func (wb *Builder) WithDescription(description string) *Builder + func (wb *Builder) WithIntermediateOutputFrom(bindings ...ExecutorBinding) *Builder + func (wb *Builder) WithName(name string) *Builder + func (wb *Builder) WithOutputFrom(bindings ...ExecutorBinding) *Builder + func (wb *Builder) WithTelemetry(tracer workflowobservability.Tracer, options TelemetryOptions) *Builder + type CatchAllFunc func(*Context, PortableValue) (any, error) + type CheckpointInfo struct + CheckpointID string + SessionID string + func NewCheckpointInfo(sessionID string) CheckpointInfo + type Context struct + AddEvent func(event Event) error + ConcurrentRunsEnabled bool + PostRequest func(request *ExternalRequest) error + QueueClearScope func(scope string) error + QueueStateUpdate func(key string, scope string, value any) error + ReadOrInitState func(key string, scope string, ...) (any, error) + ReadState func(key string, scope string) (any, error) + ReadStateKeys func(scope string) iter.Seq2[string, error] + RequestHalt func() error + SendMessage func(targetID string, message any) error + TraceContext func() map[string]any + YieldOutput func(output any) error + type Edge struct + Assigner func(int, any) iter.Seq[int] + Condition func(any) bool + Connection EdgeConnection + Index int + Label string + func (e Edge) Equal(other Edge) bool + type EdgeConnection struct + SinkIDs []string + SourceIDs []string + func (c EdgeConnection) Equal(other EdgeConnection) bool + type EdgeInfo struct + Connection EdgeConnection + HasAssigner bool + HasCondition bool + Label string + func (e *EdgeInfo) Match(other Edge) bool + type EdgeOption func(*edgeOptions) + func IdempotentEdge() EdgeOption + func WithEdgeAssigner[T any](assigner func(targetCount int, message T) iter.Seq[int]) EdgeOption + func WithEdgeCondition0(condition func() bool) EdgeOption + func WithEdgeCondition[T any](condition func(T) bool) EdgeOption + func WithEdgeLabel(label string) EdgeOption + type ErrorEvent struct + Error error + SubWorkflowID string + func (e ErrorEvent) Data() any + type Event interface + Data func() any + type Executor struct + AttachRuntimeFunc func(runtime any) error + AutoSendMessageHandlerResultObject *bool + AutoYieldOutputHandlerResultObject *bool + CloseFunc func(ctx context.Context) error + ConfigureProtocol func(builder *ProtocolBuilder) (*ProtocolBuilder, error) + CrossRunShareable bool + ID string + ImplementationID string + InitializeFunc func(ctx *Context) error + OnCheckpointFunc func(ctx *Context) error + OnCheckpointRestoredFunc func(ctx *Context) error + OnMessageDeliveryFinishedFunc func(ctx *Context) error + OnMessageDeliveryStartingFunc func(ctx *Context) error + ResetFunc func() error + func NewAggregatingExecutor[TInput, TAggregate any](id string, aggregator func(*TAggregate, TInput) *TAggregate) *Executor + func NewExecutor(id string, v any) *Executor + func (e *Executor) AttachRuntime(runtime any) error + func (e *Executor) Bind() ExecutorBinding + func (e *Executor) Close(ctx context.Context) error + func (e *Executor) DescribeProtocol() ProtocolDescriptor + func (e *Executor) Execute(ctx *Context, message any) (result any, err error) + func (e *Executor) Extend(executor *Executor) *Executor + func (e *Executor) Initialize(ctx *Context) error + func (e *Executor) OnCheckpoint(ctx *Context) error + func (e *Executor) OnCheckpointRestored(ctx *Context) error + func (e *Executor) OnMessageDeliveryFinished(ctx *Context) error + func (e *Executor) OnMessageDeliveryStarting(ctx *Context) error + func (e *Executor) Reset() error + func (e *Executor) SetCrossRunShareable(v bool) *Executor + type ExecutorBinding struct + ID string + ImplementationID string + NewExecutorFunc func(sessionID string) (*Executor, error) + Ports []RequestPort + RawValue any + ResetFunc func() bool + SharedInstance bool + SupportsConcurrentSharedExecution bool + func BindNewExecutorFunc(id string, fn func(sessionID string, executorID string) (*Executor, error)) ExecutorBinding + func (eb ExecutorBinding) CreateInstance(sessionID string) (*Executor, error) + func (eb ExecutorBinding) String() string + func (eb ExecutorBinding) TryReset() bool + type ExecutorCompletedEvent struct + ExecutorID string + Result any + func (e ExecutorCompletedEvent) Data() any + type ExecutorFailedEvent struct + Error error + ExecutorID string + func (e ExecutorFailedEvent) Data() any + type ExecutorInvokedEvent struct + ExecutorID string + Message any + func (e ExecutorInvokedEvent) Data() any + type ExternalRequest struct + Data PortableValue + PortInfo RequestPortInfo + RequestID string + func NewExternalRequest(id string, port RequestPort, data any) (*ExternalRequest, error) + func (r *ExternalRequest) CreateResponse(data any) (*ExternalResponse, error) + type ExternalResponse struct + Data PortableValue + PortInfo RequestPortInfo + RequestID string + type MessageHandlerFunc func(*Context, any) (any, error) + type OutputEvent struct + ExecutorID string + Output any + Tags []OutputTag + func (e OutputEvent) Data() any + func (e OutputEvent) HasTag(tag OutputTag) bool + func (e OutputEvent) IsIntermediate() bool + type OutputTag string + const OutputTagIntermediate + func (t *OutputTag) UnmarshalJSON(data []byte) error + func (t OutputTag) MarshalJSON() ([]byte, error) + func (t OutputTag) String() string + func (t OutputTag) Value() string + type PortableValue struct + TypeID TypeID + func AnyPortableValue(v any) PortableValue + func (v *PortableValue) Any() any + func (v *PortableValue) As(typ reflect.Type) (any, bool) + func (v *PortableValue) Delayed() bool + func (v *PortableValue) Is(typ reflect.Type) bool + func (v *PortableValue) UnmarshalJSON(data []byte) error + func (v PortableValue) MarshalJSON() ([]byte, error) + type ProtocolBuilder struct + RouteBuilder RouteBuilder + func (pb *ProtocolBuilder) ConfigureRoutes(configure func(*RouteBuilder) (*RouteBuilder, error)) *ProtocolBuilder + func (pb *ProtocolBuilder) SendsMessageType(messageTypes ...reflect.Type) *ProtocolBuilder + func (pb *ProtocolBuilder) YieldsOutputType(outputTypes ...reflect.Type) *ProtocolBuilder + type ProtocolDescriptor struct + Accepts []reflect.Type + AcceptsAll bool + Sends []reflect.Type + Yields []reflect.Type + type RequestHaltEvent struct + Result any + func (e RequestHaltEvent) Data() any + type RequestInfoEvent struct + Request *ExternalRequest + func (e RequestInfoEvent) Data() any + type RequestPort struct + ID string + Request reflect.Type + Response reflect.Type + func (p RequestPort) Bind() ExecutorBinding + type RequestPortInfo struct + PortID string + RequestType TypeID + ResponseType TypeID + func NewRequestPortInfo(port RequestPort) RequestPortInfo + type RouteBuilder struct + func (rb *RouteBuilder) AddCatchAll(handler func(*Context, PortableValue) (any, error), ...) *RouteBuilder + func (rb *RouteBuilder) AddHandlerRaw(messageType reflect.Type, outputType reflect.Type, handler MessageHandlerFunc, ...) *RouteBuilder + type ScopeID struct + ExecutorID string + ScopeName string + func (s *ScopeID) UnmarshalJSON(data []byte) error + func (s ScopeID) Equal(other ScopeID) bool + func (s ScopeID) Hash(h *maphash.Hash) + func (s ScopeID) MarshalJSON() ([]byte, error) + type ScopeKey struct + ID ScopeID + Key string + func (s ScopeKey) Equal(other ScopeKey) bool + func (s ScopeKey) Hash(h *maphash.Hash) + type StartedEvent struct + Message any + func (e StartedEvent) Data() any + type StatefulExecutorCache struct + InitialStateFactory func() T + ScopeName string + StateKey string + func (s *StatefulExecutorCache[T]) InvokeWithState(ctx *Context, skipCache bool, fn func(ctx *Context, state T) (T, error)) error + func (s *StatefulExecutorCache[T]) OnCheckpointRestored(*Context) error + func (s *StatefulExecutorCache[T]) QueueStateUpdate(ctx *Context, state T) error + func (s *StatefulExecutorCache[T]) ReadState(ctx *Context, skipCache bool) (T, error) + func (s *StatefulExecutorCache[T]) Reset() error + type SuperStepCompletedEvent struct + CompletionInfo *SuperStepCompletionInfo + StepNumber int + func (e SuperStepCompletedEvent) Data() any + type SuperStepCompletionInfo struct + ActivatedExecutors []string + CheckpointInfo *CheckpointInfo + HasPendingMessages bool + HasPendingRequests bool + InstantiatedExecutors []string + StateUpdated bool + type SuperStepStartInfo struct + HasExternalMessages bool + SendingExecutors []string + type SuperStepStartedEvent struct + StartInfo *SuperStepStartInfo + StepNumber int + func (e SuperStepStartedEvent) Data() any + type SwitchBuilder struct + func (s *SwitchBuilder) AddCase(predicate func(msg any) bool, targets ...ExecutorBinding) *SwitchBuilder + func (s *SwitchBuilder) AddToBuilder(wb *Builder) *Builder + func (s *SwitchBuilder) WithDefault(targets ...ExecutorBinding) *SwitchBuilder + type TelemetryOptions struct + DisableEdgeGroupProcess bool + DisableExecutorProcess bool + DisableMessageSend bool + DisableWorkflowBuild bool + DisableWorkflowRun bool + EnableSensitiveData bool + type TurnToken struct + EmitEvents *bool + func (t TurnToken) EmitEventsOr(defaultValue bool) bool + type TypeID struct + PackageName string + TypeName string + func NewTypeID(typ reflect.Type) TypeID + func (t TypeID) Match(typ reflect.Type) bool + func (t TypeID) MatchPolymorphic(typ reflect.Type) bool + func (t TypeID) String() string + type Workflow struct + func (w *Workflow) AllowConcurrent() bool + func (w *Workflow) CheckOwnership(token any) bool + func (w *Workflow) ContextWithTelemetry(ctx context.Context) context.Context + func (w *Workflow) DescribeProtocol() (ProtocolDescriptor, error) + func (w *Workflow) Description() string + func (w *Workflow) Edges() map[string][]Edge + func (w *Workflow) ExecutorBinding(id string) (ExecutorBinding, bool) + func (w *Workflow) HasOutputExecutor(executorID string) bool + func (w *Workflow) HasResettableExecutors() bool + func (w *Workflow) Name() string + func (w *Workflow) OutgoingEdges(executorID string) []Edge + func (w *Workflow) OutputExecutorIDs() []string + func (w *Workflow) OutputExecutors() map[string][]OutputTag + func (w *Workflow) ReflectEdges() map[string][]EdgeInfo + func (w *Workflow) ReflectExecutors() map[string]ExecutorBinding + func (w *Workflow) ReflectPorts() map[string]RequestPortInfo + func (w *Workflow) ReleaseOwnership(token any) error + func (w *Workflow) ReleaseOwnershipTo(token any, targetToken any) error + func (w *Workflow) RequestPort(id string) (RequestPort, bool) + func (w *Workflow) RequestPorts() map[string]RequestPort + func (w *Workflow) StartExecutorID() string + func (w *Workflow) TakeOwnership(token any, newToken any, subworkflow bool) error + func (w *Workflow) TryReset() bool
Go to main page