Versions in this module Expand all Collapse all v0 v0.1.0 Sep 1, 2026 Changes in this version + var ScopeIDHasher internalmaphash.Hasher[workflow.ScopeID] = scopeIDHasher + var ScopeKeyHasher internalmaphash.Hasher[workflow.ScopeKey] = scopeKeyHasher + var UpdateKeyHasher internalmaphash.Hasher[UpdateKey] = updateKeyHasher + func CanHandleType(target *workflow.Executor, typ reflect.Type) bool + func CanHandleTypeID(target *workflow.Executor, typeID workflow.TypeID) bool + func CanOutputType(source *workflow.Executor, typ reflect.Type) bool + func DeclaredSendType(source *workflow.Executor, typ reflect.Type) (reflect.Type, bool) + func SentRuntimeType(source *workflow.Executor, typeID workflow.TypeID) (reflect.Type, bool) + type ConcurrentEventSink struct + EventRaised []func(context.Context, any, workflow.Event) error + func (s *ConcurrentEventSink) AddHandler(handler func(context.Context, any, workflow.Event) error) + func (s *ConcurrentEventSink) Enqueue(ctx context.Context, evt workflow.Event) error + func (s *ConcurrentEventSink) HandlerCount() int + func (s *ConcurrentEventSink) RemoveHandler(handler func(context.Context, any, workflow.Event) error) + type DeliveryMapping struct + Envelopes []*MessageEnvelope + Targets []*workflow.Executor + func (d DeliveryMapping) MapInto(nextStep *StepContext) + type EdgeRunner struct + func NewEdgeRunner(wf *workflow.Workflow, tracer StepTracer, ...) *EdgeRunner + func (em *EdgeRunner) ExportState() (map[string]workflow.PortableValue, error) + func (em *EdgeRunner) ImportState(cp *checkpoint.Checkpoint) error + func (em *EdgeRunner) PrepareDeliveryForEdge(ctx context.Context, edge workflow.Edge, envelope *MessageEnvelope) (mapping *DeliveryMapping, err error) + func (em *EdgeRunner) PrepareDeliveryForInput(ctx context.Context, envelope *MessageEnvelope) (mapping *DeliveryMapping, err error) + func (em *EdgeRunner) PrepareDeliveryForResponse(ctx context.Context, response *workflow.ExternalResponse, ownerID string) (mapping *DeliveryMapping, err error) + type EventSink interface + Enqueue func(context.Context, workflow.Event) error + type MessageEnvelope struct + Message any + SourceID string + TargetID string + TraceContext map[string]string + func NewMessageEnvelope(message any, declaredType reflect.Type, sourceID, targetID string) (*MessageEnvelope, error) + func NewMessageEnvelopeFromPortable(envelope *checkpoint.PortableMessageEnvelope) *MessageEnvelope + func (e *MessageEnvelope) IsExternal() bool + func (e *MessageEnvelope) MessageType() workflow.TypeID + func (e *MessageEnvelope) Portable() *checkpoint.PortableMessageEnvelope + type Mode int + const ModeLockstep + const ModeOffThread + const ModeSubworkflow + type RunEventStream interface + GetStatus func(ctx context.Context) (RunStatus, error) + SignalInput func() + Start func() + Stop func() + TakeEventStream func(ctx context.Context, blockOnPendingRequest bool) iter.Seq2[workflow.Event, error] + type RunHandle struct + func NewRunHandle(sr SuperStepRunner, ch checkpoint.CheckpointingHandle, mode Mode) *RunHandle + func (h *RunHandle) Cancel() + func (h *RunHandle) Checkpoints() []workflow.CheckpointInfo + func (h *RunHandle) Close(ctx context.Context) error + func (h *RunHandle) EnqueueMessageUntyped(ctx context.Context, message any, declaredType reflect.Type) (bool, error) + func (h *RunHandle) EnqueueResponse(ctx context.Context, response *workflow.ExternalResponse) error + func (h *RunHandle) GetStatus(ctx context.Context) (RunStatus, error) + func (h *RunHandle) IsCheckpointingEnabled() bool + func (h *RunHandle) IsValidInputType(ctx context.Context, typ reflect.Type) (bool, error) + func (h *RunHandle) LastCheckpoint() (workflow.CheckpointInfo, bool) + func (h *RunHandle) ResponsePortExecutorID(portID string) (string, bool) + func (h *RunHandle) RestoreCheckpoint(ctx context.Context, checkpointInfo workflow.CheckpointInfo) error + func (h *RunHandle) SessionID() string + func (h *RunHandle) TakeEventStream(ctx context.Context, blockOnPendingRequest bool) iter.Seq2[workflow.Event, error] + type RunStatus int + const RunStatusEnded + const RunStatusIdle + const RunStatusNotStarted + const RunStatusPendingRequests + const RunStatusRunning + type StateManager struct + func NewStateManager() StateManager + func (sm *StateManager) ClearState(executorID string, scopeName string) error + func (sm *StateManager) ClearStateByID(scopeID workflow.ScopeID) error + func (sm *StateManager) ClearStateKey(executorID string, scopeName string, key string) error + func (sm *StateManager) ClearStateKeyByID(scopeID workflow.ScopeID, key string) error + func (sm *StateManager) ExportState() (iter.Seq2[workflow.ScopeKey, workflow.PortableValue], error) + func (sm *StateManager) ImportState(cp *checkpoint.Checkpoint) error + func (sm *StateManager) PublishUpdates(tracer StepTracer) error + func (sm *StateManager) ReadKeys(executorID string, scopeName string) map[string]struct{} + func (sm *StateManager) ReadKeysByID(scopeID workflow.ScopeID) map[string]struct{} + func (sm *StateManager) ReadOrInitState(executorID string, scopeName string, key string, factory func() any) (workflow.PortableValue, error) + func (sm *StateManager) ReadOrInitStateByID(scopeID workflow.ScopeID, key string, factory func() any) (workflow.PortableValue, error) + func (sm *StateManager) ReadState(executorID string, scopeName string, key string) (workflow.PortableValue, bool, error) + func (sm *StateManager) ReadStateByID(scopeID workflow.ScopeID, key string) (workflow.PortableValue, bool, error) + func (sm *StateManager) WriteState(executorID string, scopeName string, key string, value any) error + func (sm *StateManager) WriteStateByID(scopeID workflow.ScopeID, key string, value any) error + type StateScope struct + func NewStateScope(scopeID workflow.ScopeID) *StateScope + func (s *StateScope) ContainsKey(key string) bool + func (s *StateScope) ExportStates() iter.Seq2[string, workflow.PortableValue] + func (s *StateScope) ImportState(key string, state workflow.PortableValue) + func (s *StateScope) ReadKeys() map[string]struct{} + func (s *StateScope) ReadState(key string) (workflow.PortableValue, bool) + func (s *StateScope) ScopeID() workflow.ScopeID + func (s *StateScope) WriteState(updates map[string][]StateUpdate) error + type StateUpdate struct + IsDelete bool + Key string + Value any + func DeleteStateUpdate(key string) StateUpdate + func UpdateStateUpdate(key string, value any) StateUpdate + type StepContext struct + func (s *StepContext) ExportMessages() map[string][]*checkpoint.PortableMessageEnvelope + func (s *StepContext) HasMessages() bool + func (s *StepContext) ImportMessages(messages map[string][]*checkpoint.PortableMessageEnvelope) + func (s *StepContext) Keys() []string + func (s *StepContext) MessagesFor(target string) *concurrent.Queue[*MessageEnvelope] + type StepTracer interface + TraceActivated func(executorID string) + TraceCheckpointCreated func(workflow.CheckpointInfo) + TraceInstantiated func(executorID string) + TraceStatePublished func() + type SuperStepRunner interface + EnqueueMessageUntyped func(context.Context, any, reflect.Type) (bool, error) + EnqueueResponse func(context.Context, *workflow.ExternalResponse) error + HasUnprocessedMessages func() bool + HasUnservicedRequests func() bool + IsValidInputType func(context.Context, reflect.Type) (bool, error) + OutgoingEvents func() *ConcurrentEventSink + RepublishPendingEvents func(context.Context) error + RequestEndRun func(context.Context) error + ResponsePortExecutorID func(portID string) (string, bool) + RunSuperStep func(context.Context) (bool, error) + SessionID func() string + StartExecutorID func() string + Workflow func() *workflow.Workflow + type UpdateKey struct + Key string + ScopeID workflow.ScopeID + func (s UpdateKey) Hash(h *maphash.Hash) + func (u UpdateKey) Equal(other UpdateKey) bool + func (u UpdateKey) IsMatchingScope(scopeID workflow.ScopeID, strict bool) bool
Go to main page