jobqueue

Documentation

Index

• type Consumer
  • func NewConsumer(log zerolog.Logger, jobs module.Jobs, progress storage.ConsumerProgress, ...) *Consumer
  • func (c *Consumer) Check()
  • func (c *Consumer) NotifyJobIsDone(jobID module.JobID) uint64
  • func (c *Consumer) Size() uint
  • func (c *Consumer) Start(defaultIndex uint64) error
  • func (c *Consumer) Stop()
• type Worker

Types

type Consumer

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

func NewConsumer

func NewConsumer(
	log zerolog.Logger,
	jobs module.Jobs,
	progress storage.ConsumerProgress,
	worker Worker,
	maxProcessing uint64,
) *Consumer

func (*Consumer) Check

func (c *Consumer) Check()

Check allows the job publisher to notify the consumer that a new job has been added, so that the consumer can check if the job is processable since multiple checks at the same time are unnecessary, we could only keep one check by checking. an atomic isChecking value.

func (*Consumer) NotifyJobIsDone

func (c *Consumer) NotifyJobIsDone(jobID module.JobID) uint64

NotifyJobIsDone let the consumer know a job has been finished, so that consumer will take the next job from the job queue if there are workers available. It returns the last processed job index.

func (*Consumer) Size

func (c *Consumer) Size() uint

Size returns number of in-memory jobs that consumer is processing.

func (*Consumer) Start

func (c *Consumer) Start(defaultIndex uint64) error

Start starts consuming the jobs from the job queue.

func (*Consumer) Stop

func (c *Consumer) Stop()

Stop stops consuming jobs from the job queue. It blocks until the existing worker finish processing the job Note, it won't stop the existing worker from finishing their job

type Worker

type Worker interface {
	// returned error must be unexpected fatal error
	Run(job module.Job) error
}