Version: v0.19.2 Latest Latest

This package is not in the latest version of its module.

Go to latest
Published: Jul 20, 2021 License: AGPL-3.0 Imports: 7 Imported by: 0




This section is empty.


This section is empty.


This section is empty.


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

Source Files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to