batcher

Documentation

Index

• Constants
• Variables
• func NewResult(dc *dynamicconfig.Collection, params activityDeps) fxResult
• type BatchParams
• type CancelParams
• type HeartBeatDetails
  • func BatchWorkflow(ctx workflow.Context, batchParams BatchParams) (HeartBeatDetails, error)
• type SignalParams
• type TerminateParams

Constants

const (
	// BatchWFTypeName is the workflow type
	BatchWFTypeName = "temporal-sys-batch-workflow"
	// DefaultRPS is the default RPS
	DefaultRPS = 50
	// DefaultConcurrency is the default concurrency
	DefaultConcurrency = 5
)

const (
	// InfiniteDuration is a long duration(20 yrs) we used for infinite workflow running
	InfiniteDuration = 20 * 365 * 24 * time.Hour

	// DefaultAttemptsOnRetryableError is the default value for AttemptsOnRetryableError
	DefaultAttemptsOnRetryableError = 50
	// DefaultActivityHeartBeatTimeout is the default value for ActivityHeartBeatTimeout
	DefaultActivityHeartBeatTimeout = time.Second * 10
)

const (
	// BatchTypeTerminate is batch type for terminating workflows
	BatchTypeTerminate = "terminate"
	// BatchTypeCancel is the batch type for canceling workflows
	BatchTypeCancel = "cancel"
	// BatchTypeSignal is batch type for signaling workflows
	BatchTypeSignal = "signal"
)

Variables

var Module = fx.Options(
	fx.Provide(NewResult),
)

Functions

func NewResult

func NewResult(
	dc *dynamicconfig.Collection,
	params activityDeps,
) fxResult

Types

type BatchParams

type BatchParams struct {
	// Target namespace to execute batch operation
	Namespace string
	// To get the target workflows for processing
	Query string
	// Reason for the operation
	Reason string
	// Supporting: signal,cancel,terminate
	BatchType string

	// Below are all optional
	// TerminateParams is params only for BatchTypeTerminate
	TerminateParams TerminateParams
	// CancelParams is params only for BatchTypeCancel
	CancelParams CancelParams
	// SignalParams is params only for BatchTypeSignal
	SignalParams SignalParams
	// RPS of processing. Default to DefaultRPS
	// This is moving to dynamic config.
	// TODO: Remove it from BatchParams after 1.19+
	RPS int
	// Number of goroutines running in parallel to process
	// This is moving to dynamic config.
	// TODO: Remove it from BatchParams after 1.19+
	Concurrency int
	// Number of attempts for each workflow to process in case of retryable error before giving up
	AttemptsOnRetryableError int
	// timeout for activity heartbeat
	ActivityHeartBeatTimeout time.Duration
	// errors that will not retry which consumes AttemptsOnRetryableError. Default to empty
	NonRetryableErrors []string
	// contains filtered or unexported fields
}

BatchParams is the parameters for batch operation workflow

type CancelParams

type CancelParams struct {
}

CancelParams is the parameters for canceling workflow

type HeartBeatDetails

type HeartBeatDetails struct {
	PageToken   []byte
	CurrentPage int
	// This is just an estimation for visibility
	TotalEstimate int64
	// Number of workflows processed successfully
	SuccessCount int
	// Number of workflows that give up due to errors.
	ErrorCount int
}

HeartBeatDetails is the struct for heartbeat details

func BatchWorkflow

func BatchWorkflow(ctx workflow.Context, batchParams BatchParams) (HeartBeatDetails, error)

BatchWorkflow is the workflow that runs a batch job of resetting workflows

type SignalParams

type SignalParams struct {
	SignalName string
	Input      *commonpb.Payloads
}

SignalParams is the parameters for signaling workflow

type TerminateParams

type TerminateParams struct {
}

TerminateParams is the parameters for terminating workflow