worker

Documentation

Index

• Constants
• type JobClient
• type JobHandler
• type JobWorker
• type JobWorkerBuilder
  • func (builder *JobWorkerBuilder) Concurrency(concurrency int) JobWorkerBuilderStep3
  • func (builder *JobWorkerBuilder) FetchVariables(fetchVariables ...string) JobWorkerBuilderStep3
  • func (builder *JobWorkerBuilder) Handler(handler JobHandler) JobWorkerBuilderStep3
  • func (builder *JobWorkerBuilder) JobType(jobType string) JobWorkerBuilderStep2
  • func (builder *JobWorkerBuilder) MaxJobsActive(maxJobsActive int) JobWorkerBuilderStep3
  • func (builder *JobWorkerBuilder) Metrics(metrics JobWorkerMetrics) JobWorkerBuilderStep3
  • func (builder *JobWorkerBuilder) Name(name string) JobWorkerBuilderStep3
  • func (builder *JobWorkerBuilder) Open() JobWorker
  • func (builder *JobWorkerBuilder) PollInterval(pollInterval time.Duration) JobWorkerBuilderStep3
  • func (builder *JobWorkerBuilder) PollThreshold(pollThreshold float64) JobWorkerBuilderStep3
  • func (builder *JobWorkerBuilder) RequestTimeout(timeout time.Duration) JobWorkerBuilderStep3
  • func (builder *JobWorkerBuilder) Timeout(timeout time.Duration) JobWorkerBuilderStep3
• type JobWorkerBuilderStep1
  • func NewJobWorkerBuilder(gatewayClient pb.GatewayClient, jobClient JobClient, ...) JobWorkerBuilderStep1
• type JobWorkerBuilderStep2
• type JobWorkerBuilderStep3
• type JobWorkerMetrics

Constants

const (
	DefaultJobWorkerMaxJobActive  = 32
	DefaultJobWorkerConcurrency   = 4
	DefaultJobWorkerPollInterval  = 100 * time.Millisecond
	DefaultJobWorkerPollThreshold = 0.3
	RequestTimeoutOffset          = 10 * time.Second
	DefaultRequestTimeout         = 10 * time.Second
)

Types

type JobClient

type JobClient interface {
	NewCompleteJobCommand() commands.CompleteJobCommandStep1
	NewFailJobCommand() commands.FailJobCommandStep1
	NewThrowErrorCommand() commands.ThrowErrorCommandStep1
}

type JobHandler

type JobHandler func(client JobClient, job entities.Job)

type JobWorker

type JobWorker interface {
	// Initiate graceful shutdown and awaits termination
	Close()
	// Await termination of worker
	AwaitClose()
}

type JobWorkerBuilder

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

func (*JobWorkerBuilder) Concurrency

func (builder *JobWorkerBuilder) Concurrency(concurrency int) JobWorkerBuilderStep3

func (*JobWorkerBuilder) FetchVariables

func (builder *JobWorkerBuilder) FetchVariables(fetchVariables ...string) JobWorkerBuilderStep3

func (*JobWorkerBuilder) Handler

func (builder *JobWorkerBuilder) Handler(handler JobHandler) JobWorkerBuilderStep3

func (*JobWorkerBuilder) JobType

func (builder *JobWorkerBuilder) JobType(jobType string) JobWorkerBuilderStep2

func (*JobWorkerBuilder) MaxJobsActive

func (builder *JobWorkerBuilder) MaxJobsActive(maxJobsActive int) JobWorkerBuilderStep3

func (*JobWorkerBuilder) Metrics

func (builder *JobWorkerBuilder) Metrics(metrics JobWorkerMetrics) JobWorkerBuilderStep3

func (*JobWorkerBuilder) Name

func (builder *JobWorkerBuilder) Name(name string) JobWorkerBuilderStep3

func (*JobWorkerBuilder) Open

func (builder *JobWorkerBuilder) Open() JobWorker

func (*JobWorkerBuilder) PollInterval

func (builder *JobWorkerBuilder) PollInterval(pollInterval time.Duration) JobWorkerBuilderStep3

func (*JobWorkerBuilder) PollThreshold

func (builder *JobWorkerBuilder) PollThreshold(pollThreshold float64) JobWorkerBuilderStep3

func (*JobWorkerBuilder) RequestTimeout

func (builder *JobWorkerBuilder) RequestTimeout(timeout time.Duration) JobWorkerBuilderStep3

func (*JobWorkerBuilder) Timeout

func (builder *JobWorkerBuilder) Timeout(timeout time.Duration) JobWorkerBuilderStep3

type JobWorkerBuilderStep1

type JobWorkerBuilderStep1 interface {
	// Set the type of jobs to work on
	JobType(string) JobWorkerBuilderStep2
}

func NewJobWorkerBuilder

func NewJobWorkerBuilder(gatewayClient pb.GatewayClient, jobClient JobClient, retryPred func(ctx context.Context, err error) bool) JobWorkerBuilderStep1

NewJobWorkerBuilder should use the same retryPredicate used by the CredentialProvider (ShouldRetry method):

credsProvider, _ := zbc.NewOAuthCredentialsProvider(...)
worker.NewJobWorkerBuilder(..., credsProvider.ShouldRetry)

type JobWorkerBuilderStep2

type JobWorkerBuilderStep2 interface {
	// Set the handler to process jobs. The worker should complete or fail the job. The handler implementation
	// must be thread-safe.
	Handler(JobHandler) JobWorkerBuilderStep3
}

type JobWorkerBuilderStep3

type JobWorkerBuilderStep3 interface {
	// Set the name of the worker owner
	Name(string) JobWorkerBuilderStep3
	// Set the duration no other worker should work on job activated by this worker
	Timeout(time.Duration) JobWorkerBuilderStep3
	// Set the timeout for the request
	RequestTimeout(time.Duration) JobWorkerBuilderStep3
	// Set the maximum number of jobs which will be activated for this worker at the
	// same time.
	MaxJobsActive(int) JobWorkerBuilderStep3
	// Set the maximum number of concurrent spawned goroutines to complete jobs
	Concurrency(int) JobWorkerBuilderStep3
	// Set the maximal interval between polling for new jobs
	PollInterval(time.Duration) JobWorkerBuilderStep3
	// Set the threshold of buffered activated jobs before polling for new jobs, i.e. threshold * MaxJobsActive(int)
	PollThreshold(float64) JobWorkerBuilderStep3
	// Set list of variable names which should be fetched on job activation
	FetchVariables(...string) JobWorkerBuilderStep3
	// Set implementation for metrics reporting
	Metrics(metrics JobWorkerMetrics) JobWorkerBuilderStep3
	// Open the job worker and start polling and handling jobs
	Open() JobWorker
}

type JobWorkerMetrics

type JobWorkerMetrics interface {
	// Set the remaining count of scheduled jobs for a specific job
	SetJobsRemainingCount(jobType string, count int)
}