backend

Jun 28, 2024

Changes in this version

type ActivityWorkItem

+ func (wi ActivityWorkItem) IsWorkItem() bool

+ func (wi ActivityWorkItem) String() string

type Executor

+ Shutdown func(ctx context.Context) error

type OrchestrationWorkItem

+ func (wi OrchestrationWorkItem) IsWorkItem() bool

+ func (wi OrchestrationWorkItem) String() string

type WorkItem

+ IsWorkItem func() bool

Dec 18, 2023

Changes in this version

+ type OrchestrationIdReusePolicyOptions func(*protos.OrchestrationIdReusePolicy) error

+ func WithOrchestrationIdReusePolicy(policy *protos.OrchestrationIdReusePolicy) OrchestrationIdReusePolicyOptions

Mar 30, 2023

Changes in this version

type Backend

+ PurgeOrchestrationState func(context.Context, api.InstanceID) error

type TaskHubClient

+ PurgeOrchestrationState func(ctx context.Context, id api.InstanceID) error

Feb 23, 2023

Changes in this version

type TaskHubClient

+ ResumeOrchestration func(ctx context.Context, id api.InstanceID, reason string) error

+ SuspendOrchestration func(ctx context.Context, id api.InstanceID, reason string) error

Jan 25, 2023

Changes in this version

+ func WithStreamShutdownChannel(c <-chan any) grpcExecutorOptions

Jan 20, 2023

Changes in this version

+ var ErrBackendAlreadyStarted = errors.New("backend is already started")

+ func IsDurableTaskGrpcRequest(fullMethodName string) bool

+ func MarshalHistoryEvent(e *HistoryEvent) ([]byte, error)

+ func WithOnGetWorkItemsConnectionCallback(callback func(context.Context)) grpcExecutorOptions

type ActivityWorkItem

+ Properties map[string]interface{}

+ type HistoryEvent = protos.HistoryEvent

+ func UnmarshalHistoryEvent(bytes []byte) (*HistoryEvent, error)

+ type Logger interface

+ Debug func(v ...any)

+ Debugf func(format string, v ...any)

+ Error func(v ...any)

+ Errorf func(format string, v ...any)

+ Info func(v ...any)

+ Infof func(format string, v ...any)

+ Warn func(v ...any)

+ Warnf func(format string, v ...any)

+ func DefaultLogger() Logger

+ type NewTaskWorkerOptions func(*WorkerOptions)

+ func WithMaxParallelism(n int32) NewTaskWorkerOptions

type OrchestrationRuntimeState

+ func (s *OrchestrationRuntimeState) LastUpdatedTime() (time.Time, error)

type OrchestrationWorkItem

+ Properties map[string]interface{}

+ type TaskFailureDetails = protos.TaskFailureDetails

+ type TaskHubClient interface

+ FetchOrchestrationMetadata func(ctx context.Context, id api.InstanceID) (*api.OrchestrationMetadata, error)

+ RaiseEvent func(ctx context.Context, id api.InstanceID, eventName string, data any) error

+ ScheduleNewOrchestration func(ctx context.Context, orchestrator interface{}, ...) (api.InstanceID, error)

+ TerminateOrchestration func(ctx context.Context, id api.InstanceID, reason string) error

+ WaitForOrchestrationCompletion func(ctx context.Context, id api.InstanceID) (*api.OrchestrationMetadata, error)

+ WaitForOrchestrationStart func(ctx context.Context, id api.InstanceID) (*api.OrchestrationMetadata, error)

+ func NewTaskHubClient(be Backend) TaskHubClient

+ type WorkerOptions struct

+ MaxParallelWorkItems int32

+ func NewWorkerOptions() *WorkerOptions

Sep 15, 2022

Changes in this version

+ var ErrDuplicateEvent = errors.New("duplicate event")

+ var ErrNoWorkItems = errors.New("no work items were found")

+ var ErrNotInitialized = errors.New("backend not initialized")

+ var ErrTaskHubExists = errors.New("task hub already exists")

+ var ErrTaskHubNotFound = errors.New("task hub not found")

+ var ErrWorkItemLockLost = errors.New("lock on work-item was lost")

+ func FromRuntimeStatusString(status string) protos.OrchestrationStatus

+ func ToRuntimeStatusString(status protos.OrchestrationStatus) string

+ type ActivityExecutor interface

+ ExecuteActivity func(context.Context, api.InstanceID, *protos.HistoryEvent) (*protos.HistoryEvent, error)

+ type ActivityWorkItem struct

+ InstanceID api.InstanceID

+ LockedBy string

+ NewEvent *protos.HistoryEvent

+ Result *protos.HistoryEvent

+ SequenceNumber int64

+ func (wi *ActivityWorkItem) Description() string

+ type Backend interface

+ AbandonActivityWorkItem func(context.Context, *ActivityWorkItem) error

+ AbandonOrchestrationWorkItem func(context.Context, *OrchestrationWorkItem) error

+ AddNewOrchestrationEvent func(context.Context, api.InstanceID, *protos.HistoryEvent) error

+ CompleteActivityWorkItem func(context.Context, *ActivityWorkItem) error

+ CompleteOrchestrationWorkItem func(context.Context, *OrchestrationWorkItem) error

+ CreateOrchestrationInstance func(context.Context, *protos.HistoryEvent) error

+ CreateTaskHub func(context.Context) error

+ DeleteTaskHub func(context.Context) error

+ GetActivityWorkItem func(context.Context) (*ActivityWorkItem, error)

+ GetOrchestrationMetadata func(context.Context, api.InstanceID) (*api.OrchestrationMetadata, error)

+ GetOrchestrationRuntimeState func(context.Context, *OrchestrationWorkItem) (*OrchestrationRuntimeState, error)

+ GetOrchestrationWorkItem func(context.Context) (*OrchestrationWorkItem, error)

+ Start func(context.Context) error

+ Stop func(context.Context) error

+ type ExecutionResults struct

+ Response *protos.OrchestratorResponse

+ type Executor interface

+ ExecuteActivity func(context.Context, api.InstanceID, *protos.HistoryEvent) (*protos.HistoryEvent, error)

+ ExecuteOrchestrator func(ctx context.Context, iid api.InstanceID, oldEvents []*protos.HistoryEvent, ...) (*ExecutionResults, error)

+ func NewGrpcExecutor(grpcServer *grpc.Server, be Backend) Executor

+ type OrchestrationRuntimeState struct

+ CustomStatus *wrapperspb.StringValue

+ func NewOrchestrationRuntimeState(instanceID api.InstanceID, existingHistory []*protos.HistoryEvent) *OrchestrationRuntimeState

+ func (s *OrchestrationRuntimeState) AddEvent(e *protos.HistoryEvent) error

+ func (s *OrchestrationRuntimeState) ApplyActions(actions []*protos.OrchestratorAction) (bool, error)

+ func (s *OrchestrationRuntimeState) CompletedTime() (time.Time, error)

+ func (s *OrchestrationRuntimeState) ContinuedAsNew() bool

+ func (s *OrchestrationRuntimeState) CreatedTime() (time.Time, error)

+ func (s *OrchestrationRuntimeState) FailureDetails() (*protos.TaskFailureDetails, error)

+ func (s *OrchestrationRuntimeState) Input() (string, error)

+ func (s *OrchestrationRuntimeState) InstanceID() api.InstanceID

+ func (s *OrchestrationRuntimeState) IsCompleted() bool

+ func (s *OrchestrationRuntimeState) IsValid() bool

+ func (s *OrchestrationRuntimeState) Name() (string, error)

+ func (s *OrchestrationRuntimeState) NewEvents() []*protos.HistoryEvent

+ func (s *OrchestrationRuntimeState) OldEvents() []*protos.HistoryEvent

+ func (s *OrchestrationRuntimeState) Output() (string, error)

+ func (s *OrchestrationRuntimeState) PendingMessages() []OrchestratorMessage

+ func (s *OrchestrationRuntimeState) PendingTasks() []*protos.HistoryEvent

+ func (s *OrchestrationRuntimeState) PendingTimers() []*protos.HistoryEvent

+ func (s *OrchestrationRuntimeState) RuntimeStatus() protos.OrchestrationStatus

+ func (s *OrchestrationRuntimeState) String() string

+ type OrchestrationWorkItem struct

+ InstanceID api.InstanceID

+ LockedBy string

+ NewEvents []*protos.HistoryEvent

+ RetryCount int32

+ State *OrchestrationRuntimeState

+ func (wi *OrchestrationWorkItem) Description() string

+ func (wi *OrchestrationWorkItem) GetAbandonDelay() time.Duration

+ type OrchestratorExecutor interface

+ ExecuteOrchestrator func(ctx context.Context, iid api.InstanceID, oldEvents []*protos.HistoryEvent, ...) (*ExecutionResults, error)

+ type OrchestratorMessage struct

+ HistoryEvent *protos.HistoryEvent

+ TargetInstanceID string

+ type TaskHubWorker interface

+ Shutdown func(context.Context) error

+ Start func(context.Context) error

+ func NewTaskHubWorker(be Backend, orchestrationWorker TaskWorker, activityWorker TaskWorker) TaskHubWorker

+ type TaskProcessor interface

+ AbandonWorkItem func(context.Context, WorkItem) error

+ CompleteWorkItem func(context.Context, WorkItem) error

+ FetchWorkItem func(context.Context) (WorkItem, error)

+ Name func() string

+ ProcessWorkItem func(context.Context, WorkItem) error

+ type TaskWorker interface

+ ProcessNext func(context.Context) (bool, error)

+ Start func(context.Context)

+ StopAndDrain func()

+ func NewActivityTaskWorker(be Backend, executor ActivityExecutor) TaskWorker

+ func NewOrchestrationWorker(be Backend, executor OrchestratorExecutor) TaskWorker

+ func NewTaskWorker(be Backend, p TaskProcessor) TaskWorker

+ type WorkItem interface

+ Description func() string