Documentation
¶
Index ¶
- Variables
- func GetActivityExecutionKey(iid string, taskID int32) string
- func IsDurableTaskGrpcRequest(fullMethodName string) bool
- func MarshalHistoryEvent(e *HistoryEvent) ([]byte, error)
- func WithOnGetWorkItemsConnectionCallback(callback func(context.Context) error) grpcExecutorOptions
- func WithOnGetWorkItemsDisconnectCallback(callback func(context.Context) error) grpcExecutorOptions
- func WithSkipWaitForInstanceStart() grpcExecutorOptions
- func WithStreamSendTimeout(d time.Duration) grpcExecutorOptions
- func WithStreamShutdownChannel(c <-chan any) grpcExecutorOptions
- type ActivityExecutor
- type ActivityRequest
- type ActivityWorkItem
- type Backend
- type CreateWorkflowInstanceRequest
- type DurableTimer
- type Executor
- type HistoryEvent
- type Logger
- type NewTaskWorkerOptions
- type OrchestrationIdReusePolicyOptions
- type OrchestrationMetadata
- type OrchestrationRuntimeState
- type OrchestrationRuntimeStateMessage
- type OrchestrationStatus
- type OrchestrationWorkItem
- type OrchestratorExecutor
- type RerunWorkflowFromEventRequest
- type TaskFailureDetails
- type TaskHubClient
- type TaskHubWorker
- type TaskProcessor
- type TaskWorker
- func NewActivityTaskWorker(be Backend, executor ActivityExecutor, logger Logger, ...) TaskWorker[*ActivityWorkItem]
- func NewOrchestrationWorker(be Backend, executor OrchestratorExecutor, logger Logger, ...) TaskWorker[*OrchestrationWorkItem]
- func NewTaskWorker[T WorkItem](p TaskProcessor[T], logger Logger, opts ...NewTaskWorkerOptions) TaskWorker[T]
- type WorkItem
- type WorkerOptions
- type WorkflowState
- type WorkflowStateMetadata
Constants ¶
This section is empty.
Variables ¶
var ( ErrTaskHubExists = errors.New("task hub already exists") ErrTaskHubNotFound = errors.New("task hub not found") ErrNotInitialized = errors.New("backend not initialized") ErrWorkItemLockLost = errors.New("lock on work-item was lost") ErrBackendAlreadyStarted = errors.New("backend is already started") )
Functions ¶
func GetActivityExecutionKey ¶ added in v0.8.0
func IsDurableTaskGrpcRequest ¶
IsDurableTaskGrpcRequest returns true if the specified gRPC method name represents an operation that is compatible with the gRPC executor.
func MarshalHistoryEvent ¶
func MarshalHistoryEvent(e *HistoryEvent) ([]byte, error)
MarshalHistoryEvent serializes the HistoryEvent into a protobuf byte array.
func WithOnGetWorkItemsConnectionCallback ¶
WithOnGetWorkItemsConnectionCallback allows the caller to get a notification when an external process connects over gRPC and invokes the GetWorkItems operation. This can be useful for doing things like lazily auto-starting the task hub worker only when necessary.
func WithOnGetWorkItemsDisconnectCallback ¶
WithOnGetWorkItemsDisconnectCallback allows the caller to get a notification when an external process disconnects from the GetWorkItems operation. This can be useful for doing things like shutting down the task hub worker when the client disconnects.
func WithSkipWaitForInstanceStart ¶ added in v0.8.0
func WithSkipWaitForInstanceStart() grpcExecutorOptions
func WithStreamSendTimeout ¶ added in v0.6.7
func WithStreamShutdownChannel ¶
func WithStreamShutdownChannel(c <-chan any) grpcExecutorOptions
Types ¶
type ActivityExecutor ¶
type ActivityExecutor interface {
ExecuteActivity(context.Context, api.InstanceID, *protos.HistoryEvent) (*protos.HistoryEvent, error)
}
type ActivityRequest ¶
type ActivityRequest = protos.ActivityRequest
type ActivityWorkItem ¶
type ActivityWorkItem struct { SequenceNumber int64 InstanceID api.InstanceID NewEvent *HistoryEvent Result *HistoryEvent LockedBy string Properties map[string]interface{} }
func (ActivityWorkItem) IsWorkItem ¶
func (wi ActivityWorkItem) IsWorkItem() bool
IsWorkItem implements core.WorkItem
func (ActivityWorkItem) String ¶
func (wi ActivityWorkItem) String() string
String implements core.WorkItem and fmt.Stringer
type Backend ¶
type Backend interface { // CreateTaskHub creates a new task hub for the current backend. Task hub creation must be idempotent. // // If the task hub for this backend already exists, an error of type [ErrTaskHubExists] is returned. CreateTaskHub(context.Context) error // DeleteTaskHub deletes an existing task hub configured for the current backend. It's up to the backend // implementation to determine how the task hub data is deleted. // // If the task hub for this backend doesn't exist, an error of type [ErrTaskHubNotFound] is returned. DeleteTaskHub(context.Context) error // Start starts any background processing done by this backend. Start(context.Context) error // Stop stops any background processing done by this backend. Stop(context.Context) error // CreateOrchestrationInstance creates a new orchestration instance with a history event that // wraps a ExecutionStarted event. CreateOrchestrationInstance(context.Context, *HistoryEvent, ...OrchestrationIdReusePolicyOptions) error // RerunWorkflowFromEvent reruns a workflow from a specific event ID of some // source instance ID. If not given, a random new instance ID will be // generated and returned. Can optionally give a new input to the target // event ID to rerun from. RerunWorkflowFromEvent(ctx context.Context, req *protos.RerunWorkflowFromEventRequest) (api.InstanceID, error) // AddNewEvent adds a new orchestration event to the specified orchestration instance. AddNewOrchestrationEvent(context.Context, api.InstanceID, *HistoryEvent) error // NextOrchestrationWorkItem blocks and returns the next orchestration work // item from the task hub. Should only return an error when shutting down. NextOrchestrationWorkItem(context.Context) (*OrchestrationWorkItem, error) // GetOrchestrationRuntimeState gets the runtime state of an orchestration instance. GetOrchestrationRuntimeState(context.Context, *OrchestrationWorkItem) (*OrchestrationRuntimeState, error) // WatchOrchestrationRuntimeStatus is a streaming API to watch for changes to // the OrchestrtionMetadata, receiving events as and when the state changes. // When the given condition is true, returns. // Used over polling the metadata. WatchOrchestrationRuntimeStatus(ctx context.Context, id api.InstanceID, condition func(*OrchestrationMetadata) bool) error // GetOrchestrationMetadata gets the metadata associated with the given orchestration instance ID. // // Returns [api.ErrInstanceNotFound] if the orchestration instance doesn't exist. GetOrchestrationMetadata(context.Context, api.InstanceID) (*OrchestrationMetadata, error) // CompleteOrchestrationWorkItem completes a work item by saving the updated runtime state to durable storage. // // Returns [ErrWorkItemLockLost] if the work-item couldn't be completed due to a lock-lost conflict (e.g., split-brain). CompleteOrchestrationWorkItem(context.Context, *OrchestrationWorkItem) error // AbandonOrchestrationWorkItem undoes any state changes and returns the work item to the work item queue. // // This is called if an internal failure happens in the processing of an orchestration work item. It is // not called if the orchestration work item is processed successfully (note that an orchestration that // completes with a failure is still considered a successfully processed work item). AbandonOrchestrationWorkItem(context.Context, *OrchestrationWorkItem) error // NextActivityWorkItem blocks and returns the next activity work item from // the task hub. Should only return an error when shutting down. NextActivityWorkItem(context.Context) (*ActivityWorkItem, error) // CompleteActivityWorkItem sends a message to the parent orchestration indicating activity completion. // // Returns [ErrWorkItemLockLost] if the work-item couldn't be completed due to a lock-lost conflict (e.g., split-brain). CompleteActivityWorkItem(context.Context, *ActivityWorkItem) error // AbandonActivityWorkItem returns the work-item back to the queue without committing any other chances. // // This is called when an internal failure occurs during activity work-item processing. AbandonActivityWorkItem(context.Context, *ActivityWorkItem) error // PurgeOrchestrationState deletes all saved state for the specified orchestration instance. // // [api.ErrInstanceNotFound] is returned if the specified orchestration instance doesn't exist. // [api.ErrNotCompleted] is returned if the specified orchestration instance is still running. PurgeOrchestrationState(context.Context, api.InstanceID) error // CompleteOrchestratorTask completes the orchestrator task by saving the updated runtime state to durable storage. CompleteOrchestratorTask(context.Context, *protos.OrchestratorResponse) error // CancelOrchestratorTask cancels the orchestrator task so instances of WaitForOrchestratorCompletion will return an error. CancelOrchestratorTask(context.Context, api.InstanceID) error // WaitForOrchestratorCompletion blocks until the orchestrator completes and returns the final response. // // [api.ErrTaskCancelled] is returned if the task was cancelled. WaitForOrchestratorCompletion(context.Context, *protos.OrchestratorRequest) (*protos.OrchestratorResponse, error) // CompleteActivityTask completes the activity task by saving the updated runtime state to durable storage. CompleteActivityTask(context.Context, *protos.ActivityResponse) error // CancelActivityTask cancels the activity task so instances of WaitForActivityCompletion will return an error. CancelActivityTask(context.Context, api.InstanceID, int32) error // WaitForActivityCompletion blocks until the activity completes and returns the final response. // // [api.ErrTaskCancelled] is returned if the task was cancelled. WaitForActivityCompletion(context.Context, *protos.ActivityRequest) (*protos.ActivityResponse, error) }
type CreateWorkflowInstanceRequest ¶
type CreateWorkflowInstanceRequest = protos.CreateWorkflowInstanceRequest
type DurableTimer ¶
type DurableTimer = protos.DurableTimer
type Executor ¶
type Executor interface { ExecuteOrchestrator(ctx context.Context, iid api.InstanceID, oldEvents []*protos.HistoryEvent, newEvents []*protos.HistoryEvent) (*protos.OrchestratorResponse, error) ExecuteActivity(context.Context, api.InstanceID, *protos.HistoryEvent) (*protos.HistoryEvent, error) Shutdown(ctx context.Context) error }
func NewGrpcExecutor ¶
func NewGrpcExecutor(be Backend, logger Logger, opts ...grpcExecutorOptions) (executor Executor, registerServerFn func(grpcServer grpc.ServiceRegistrar))
NewGrpcExecutor returns the Executor object and a method to invoke to register the gRPC server in the executor.
type HistoryEvent ¶
type HistoryEvent = protos.HistoryEvent
func UnmarshalHistoryEvent ¶
func UnmarshalHistoryEvent(bytes []byte) (*HistoryEvent, error)
UnmarshalHistoryEvent deserializes a HistoryEvent from a protobuf byte array.
type Logger ¶
type Logger interface { // Debug logs a message at level Debug. Debug(v ...any) // Debugf logs a message at level Debug. Debugf(format string, v ...any) // Info logs a message at level Info. Info(v ...any) // Infof logs a message at level Info. Infof(format string, v ...any) // Warn logs a message at level Warn. Warn(v ...any) // Warnf logs a message at level Warn. Warnf(format string, v ...any) // Error logs a message at level Error. Error(v ...any) // Errorf logs a message at level Error. Errorf(format string, v ...any) }
func DefaultLogger ¶
func DefaultLogger() Logger
type NewTaskWorkerOptions ¶
type NewTaskWorkerOptions func(*WorkerOptions)
func WithMaxParallelism ¶
func WithMaxParallelism(n int32) NewTaskWorkerOptions
type OrchestrationIdReusePolicyOptions ¶
type OrchestrationIdReusePolicyOptions func(*protos.OrchestrationIdReusePolicy) error
func WithOrchestrationIdReusePolicy ¶
func WithOrchestrationIdReusePolicy(policy *protos.OrchestrationIdReusePolicy) OrchestrationIdReusePolicyOptions
type OrchestrationMetadata ¶
type OrchestrationMetadata = protos.OrchestrationMetadata
type OrchestrationRuntimeState ¶
type OrchestrationRuntimeState = protos.OrchestrationRuntimeState
type OrchestrationRuntimeStateMessage ¶
type OrchestrationRuntimeStateMessage = protos.OrchestrationRuntimeStateMessage
type OrchestrationStatus ¶
type OrchestrationStatus = protos.OrchestrationStatus
type OrchestrationWorkItem ¶
type OrchestrationWorkItem struct { InstanceID api.InstanceID NewEvents []*HistoryEvent LockedBy string RetryCount int32 State *protos.OrchestrationRuntimeState Properties map[string]interface{} }
func (*OrchestrationWorkItem) GetAbandonDelay ¶
func (wi *OrchestrationWorkItem) GetAbandonDelay() time.Duration
func (OrchestrationWorkItem) IsWorkItem ¶
func (wi OrchestrationWorkItem) IsWorkItem() bool
IsWorkItem implements core.WorkItem
func (OrchestrationWorkItem) String ¶
func (wi OrchestrationWorkItem) String() string
String implements core.WorkItem and fmt.Stringer
type OrchestratorExecutor ¶
type OrchestratorExecutor interface { ExecuteOrchestrator( ctx context.Context, iid api.InstanceID, oldEvents []*protos.HistoryEvent, newEvents []*protos.HistoryEvent) (*protos.OrchestratorResponse, error) }
type RerunWorkflowFromEventRequest ¶ added in v0.7.0
type RerunWorkflowFromEventRequest = protos.RerunWorkflowFromEventRequest
type TaskFailureDetails ¶
type TaskFailureDetails = protos.TaskFailureDetails
type TaskHubClient ¶
type TaskHubClient interface { ScheduleNewOrchestration(ctx context.Context, orchestrator interface{}, opts ...api.NewOrchestrationOptions) (api.InstanceID, error) FetchOrchestrationMetadata(ctx context.Context, id api.InstanceID) (*OrchestrationMetadata, error) WaitForOrchestrationStart(ctx context.Context, id api.InstanceID) (*OrchestrationMetadata, error) WaitForOrchestrationCompletion(ctx context.Context, id api.InstanceID) (*OrchestrationMetadata, error) TerminateOrchestration(ctx context.Context, id api.InstanceID, opts ...api.TerminateOptions) error RaiseEvent(ctx context.Context, id api.InstanceID, eventName string, opts ...api.RaiseEventOptions) error SuspendOrchestration(ctx context.Context, id api.InstanceID, reason string) error ResumeOrchestration(ctx context.Context, id api.InstanceID, reason string) error PurgeOrchestrationState(ctx context.Context, id api.InstanceID, opts ...api.PurgeOptions) error RerunWorkflowFromEvent(ctx context.Context, source api.InstanceID, eventID uint32, opts ...api.RerunOptions) (api.InstanceID, error) }
func NewTaskHubClient ¶
func NewTaskHubClient(be Backend) TaskHubClient
type TaskHubWorker ¶
type TaskHubWorker interface { // Start starts the backend and the configured internal workers. Start(context.Context) error // Shutdown stops the backend and all internal workers. Shutdown(context.Context) error }
func NewTaskHubWorker ¶
func NewTaskHubWorker(be Backend, orchestrationWorker TaskWorker[*OrchestrationWorkItem], activityWorker TaskWorker[*ActivityWorkItem], logger Logger) TaskHubWorker
type TaskProcessor ¶
type TaskWorker ¶
type TaskWorker[T WorkItem] interface { // Start starts background polling for the activity work items. Start(context.Context) // StopAndDrain stops the worker and waits for all outstanding work items to finish. StopAndDrain() }
func NewActivityTaskWorker ¶
func NewActivityTaskWorker(be Backend, executor ActivityExecutor, logger Logger, opts ...NewTaskWorkerOptions) TaskWorker[*ActivityWorkItem]
func NewOrchestrationWorker ¶
func NewOrchestrationWorker(be Backend, executor OrchestratorExecutor, logger Logger, opts ...NewTaskWorkerOptions) TaskWorker[*OrchestrationWorkItem]
func NewTaskWorker ¶
func NewTaskWorker[T WorkItem](p TaskProcessor[T], logger Logger, opts ...NewTaskWorkerOptions) TaskWorker[T]
type WorkerOptions ¶
type WorkerOptions struct {
MaxParallelWorkItems *int32
}
func NewWorkerOptions ¶
func NewWorkerOptions() *WorkerOptions
type WorkflowState ¶
type WorkflowState = protos.WorkflowState
type WorkflowStateMetadata ¶
type WorkflowStateMetadata = protos.WorkflowStateMetadata