aggregatedpool

Documentation

Index

• Constants
• func NewWorkerInterceptor() interceptor.WorkerInterceptor
• func TemporalWorkers(wDef *Workflow, actDef *Activity, wi []*internal.WorkerInfo, log *zap.Logger, ...) ([]worker.Worker, error)
• type Activity
  • func NewActivityDefinition(ac common.Codec, p common.Pool, log *zap.Logger) *Activity
  • func (a *Activity) GetActivityContext(taskToken []byte) (context.Context, error)
• type Callback
• type LaFn
• type LocalActivityFn
  • func NewLocalActivityFn(codec common.Codec, pool common.Pool, log *zap.Logger) *LocalActivityFn
  • func (la *LocalActivityFn) ExecuteLA(ctx context.Context, hdr *commonpb.Header, args *commonpb.Payloads) (*commonpb.Payloads, error)
• type Workflow
  • func NewWorkflowDefinition(codec common.Codec, la LaFn, pool common.Pool, log *zap.Logger) *Workflow
  • func (wp *Workflow) Close()
  • func (wp *Workflow) Execute(env bindings.WorkflowEnvironment, header *commonpb.Header, ...)
  • func (wp *Workflow) NewWorkflowDefinition() bindings.WorkflowDefinition
  • func (wp *Workflow) OnWorkflowTaskStarted(t time.Duration)
  • func (wp *Workflow) StackTrace() string

Constants

const (
	RrMetricName          string = "rr_activities_pool_queue_size"
	RrWorkflowsMetricName string = "rr_workflows_pool_queue_size"
)

Functions

func NewWorkerInterceptor

func NewWorkerInterceptor() interceptor.WorkerInterceptor

func TemporalWorkers

func TemporalWorkers(wDef *Workflow, actDef *Activity, wi []*internal.WorkerInfo, log *zap.Logger, tc temporalClient.Client, interceptors map[string]common.Interceptor) ([]worker.Worker, error)

Types

type Activity

type Activity struct {
	// contains filtered or unexported fields
}

func NewActivityDefinition

func NewActivityDefinition(ac common.Codec, p common.Pool, log *zap.Logger) *Activity

func (*Activity) GetActivityContext

func (a *Activity) GetActivityContext(taskToken []byte) (context.Context, error)

type Callback

type Callback func() error

type LaFn

type LaFn func(ctx context.Context, hdr *commonpb.Header, args *commonpb.Payloads) (*commonpb.Payloads, error)

type LocalActivityFn

type LocalActivityFn struct {
	// contains filtered or unexported fields
}

func NewLocalActivityFn

func NewLocalActivityFn(codec common.Codec, pool common.Pool, log *zap.Logger) *LocalActivityFn

func (*LocalActivityFn) ExecuteLA

func (la *LocalActivityFn) ExecuteLA(ctx context.Context, hdr *commonpb.Header, args *commonpb.Payloads) (*commonpb.Payloads, error)

type Workflow

type Workflow struct {
	// contains filtered or unexported fields
}

func NewWorkflowDefinition

func NewWorkflowDefinition(codec common.Codec, la LaFn, pool common.Pool, log *zap.Logger) *Workflow

NewWorkflowDefinition ... WorkflowDefinition Constructor

func (*Workflow) Close

func (wp *Workflow) Close()

func (*Workflow) Execute

func (wp *Workflow) Execute(env bindings.WorkflowEnvironment, header *commonpb.Header, input *commonpb.Payloads)

Execute implementation must be asynchronous.

func (*Workflow) NewWorkflowDefinition

func (wp *Workflow) NewWorkflowDefinition() bindings.WorkflowDefinition

NewWorkflowDefinition ... Workflow should match the WorkflowDefinitionFactory interface (sdk-go/internal/internal_worker.go:463, RegisterWorkflowWithOptions func) DO NOT USE THIS FUNCTION DIRECTLY!!!! This function called after the constructor above, it is safe to assign fields like that

func (*Workflow) OnWorkflowTaskStarted

func (wp *Workflow) OnWorkflowTaskStarted(t time.Duration)

OnWorkflowTaskStarted is called for each non-timed out startWorkflowTask event. Executed after all history events since the previous commands are applied to WorkflowDefinition Application level code must be executed from this function only. Execute call as well as callbacks called from WorkflowEnvironment functions can only schedule callbacks which can be executed from OnWorkflowTaskStarted(). FROM THE TEMPORAL DESCRIPTION

func (*Workflow) StackTrace

func (wp *Workflow) StackTrace() string

StackTrace of all coroutines owned by the Dispatcher instance.