batch

Documentation

Index

• Constants
• Variables
• func BatchWorkflow(ctx temporalsdk_workflow.Context, params BatchWorkflowInput) error
• func NewService(logger logr.Logger, cc temporalsdk_client.Client, taskQueue string, ...) *batchImpl
• type BatchActivity
  • func NewBatchActivity(batchsvc Service) *BatchActivity
  • func (a *BatchActivity) Execute(ctx context.Context, params BatchWorkflowInput) error
• type BatchProgress
• type BatchWorkflowInput
• type Service

Constants

const (
	BatchWorkflowName = "batch-workflow"
	BatchWorkflowID   = "batch-workflow"
	BatchActivityName = "batch-activity"
)

Variables

var ErrBatchStatusUnavailable = errors.New("batch status unavailable")

Functions

func BatchWorkflow

func BatchWorkflow(ctx temporalsdk_workflow.Context, params BatchWorkflowInput) error

func NewService

func NewService(logger logr.Logger, cc temporalsdk_client.Client, taskQueue string, completedDirs []string) *batchImpl

Types

type BatchActivity

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

func NewBatchActivity

func NewBatchActivity(batchsvc Service) *BatchActivity

func (*BatchActivity) Execute

func (a *BatchActivity) Execute(ctx context.Context, params BatchWorkflowInput) error

type BatchProgress

type BatchProgress struct {
	CurrentID uint
}

type BatchWorkflowInput

type BatchWorkflowInput struct {
	Path             string
	PipelineName     string
	ProcessingConfig string
	CompletedDir     string
	RetentionPeriod  *time.Duration
	RejectDuplicates bool
}

type Service

type Service interface {
	Submit(context.Context, *goabatch.SubmitPayload) (res *goabatch.BatchResult, err error)
	Status(context.Context) (res *goabatch.BatchStatusResult, err error)
	Hints(context.Context) (res *goabatch.BatchHintsResult, err error)
	InitProcessingWorkflow(ctx context.Context, req *collection.ProcessingWorkflowRequest) error
}