aggregatedpool

Documentation

Index

• Constants
• func Init(wDef *Workflow, actDef *Activity, p pool.Pool, c Codec, log *zap.Logger, ...) ([]worker.Worker, map[string]internal.WorkflowInfo, []string, error)
• type Activity
  • func NewActivityDefinition(ac Codec, p pool.Pool, log *zap.Logger, dc converter.DataConverter, ...) *Activity
  • func (a *Activity) GetActivityContext(taskToken []byte) (context.Context, error)
• type Callback
• type Codec
• type Workflow
  • func NewWorkflowDefinition(codec Codec, dc converter.DataConverter, pool pool.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 Init

func Init(wDef *Workflow, actDef *Activity, p pool.Pool, c Codec, log *zap.Logger, tc temporalClient.Client, graceTimeout time.Duration, version string) ([]worker.Worker, map[string]internal.WorkflowInfo, []string, error)

Types

type Activity

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

func NewActivityDefinition

func NewActivityDefinition(ac Codec, p pool.Pool, log *zap.Logger, dc converter.DataConverter, client temporalClient.Client, gt time.Duration) *Activity

func (*Activity) GetActivityContext

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

type Callback

type Callback func() error

type Codec

type Codec interface {
	// Encode encodes messages and context to the payload for the worker
	Encode(ctx *internal.Context, p *payload.Payload, msg ...*internal.Message) error
	// Decode decodes payload from the worker to the proto-message
	Decode(pld *payload.Payload, msg *[]*internal.Message) error
	// DecodeWorkerInfo decode a call to get a worker info ID=0 (initial)
	DecodeWorkerInfo(p *payload.Payload, wi *[]*internal.WorkerInfo) error
}

type Workflow

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

func NewWorkflowDefinition

func NewWorkflowDefinition(codec Codec, dc converter.DataConverter, pool pool.Pool, log *zap.Logger, seqID func() uint64, client temporalClient.Client, gt time.Duration) *Workflow

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!!!!

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.