queue

Documentation

Index

• Constants
• Variables
• type Job
  • func (job *Job) GetData(data any) (err error)
• type JobStatus
  • func (status JobStatus) MarshalText() (ret []byte, err error)
  • func (status JobStatus) String() string
  • func (status *JobStatus) UnmarshalText(data []byte) (err error)
• type NewJobInput
• type Queue
• type RetryStrategy
  • func (strategy RetryStrategy) MarshalText() (ret []byte, err error)
  • func (strategy RetryStrategy) String() string
  • func (strategy *RetryStrategy) UnmarshalText(data []byte) (err error)

Constants

const (
	MinRetryMax     int64 = 0
	MaxRetryMax     int64 = 100
	DefaultRetryMax int64 = 5

	MinRetryDelay     int64 = 1
	MaxRetryDelay     int64 = 86_400 // 1 day
	DefaultRetryDelay int64 = 5

	DefaultRetryStrategy = RetryStrategyConstant

	MinTimeout     int64 = 1
	MaxTimeout     int64 = 7200
	DefaultTimeout int64 = 60
)

Variables

var (
	ErrJobSatusIsNotValid = func(status string) error {
		return fmt.Errorf(`Job status "%s" is not valid`, status)
	}
	ErrRetryStrategyIsNotValid = func(status string) error {
		return fmt.Errorf(`Retry strategy "%s" is not valid`, status)
	}
)

Types

type Job

type Job struct {
	ID             guid.GUID `db:"id" json:"id"`
	CreatedAt      time.Time `db:"created_at" json:"created_at"`
	UpdatedAt      time.Time `db:"updated_at" json:"updated_at"`
	ScheduledFor   time.Time `db:"scheduled_for" json:"scheduled_for"`
	FailedAttempts int64     `db:"failed_attempts" json:"failed_attempts"`
	// priority: i64,
	Status        JobStatus       `db:"status" json:"status"`
	Type          string          `db:"type" json:"type"`
	RawData       json.RawMessage `db:"data" json:"data"`
	RetryMax      int64           `db:"retry_max" json:"retry_max"`
	RetryDelay    int64           `db:"retry_delay" json:"retry_delay"`
	RetryStrategy RetryStrategy   `db:"retry_strategy" json:"retry_strategy"`
	Timeout       int64           `db:"timeout" json:"timeout"`
}

func (*Job) GetData

func (job *Job) GetData(data any) (err error)

type JobStatus

type JobStatus int32

const (
	JobStatusQueued JobStatus = iota
	JobStatusRunning
	JobStatusFailed
)

func (JobStatus) MarshalText

func (status JobStatus) MarshalText() (ret []byte, err error)

func (JobStatus) String

func (status JobStatus) String() string

func (*JobStatus) UnmarshalText

func (status *JobStatus) UnmarshalText(data []byte) (err error)

UnmarshalText implements encoding.TextUnmarshaler.

type NewJobInput

type NewJobInput struct {
	Type string
	Data any

	// ScheduledFor is the date when the job should be scheduled for
	// default: time.Now()
	ScheduledFor *time.Time

	// RetryMax is the max number of times a job should be retried
	// 0-100
	// default: 5
	RetryMax *int64

	// RetryDelay is the number of seconds between 2 retry attempts. Allowed range: 1-86400
	// default: 5
	RetryDelay *int64

	// constant, exponential
	// default: Constant
	RetryStrategy RetryStrategy

	// Timeout in seconds. Allows range: 1-7200
	// default: 60
	Timeout *int64
}

type Queue

type Queue interface {
	Push(ctx context.Context, tx db.Queryer, newJob NewJobInput) error
	PushMany(ctx context.Context, newJobs []NewJobInput) error
	// pull fetches at most `number_of_jobs` from the queue.
	Pull(ctx context.Context, numberOfJobs uint64) ([]Job, error)
	DeleteJob(ctx context.Context, jobID guid.GUID) error
	FailJob(ctx context.Context, job Job) error
	Clear(ctx context.Context) error
	GetJob(ctx context.Context, jobID guid.GUID) (job Job, err error)

	GetFailedJobs(ctx context.Context) (jobs []Job, err error)
}

type RetryStrategy

type RetryStrategy int32

const (
	RetryStrategyConstant RetryStrategy = iota
	RetryStrategyExponential
)

func (RetryStrategy) MarshalText

func (strategy RetryStrategy) MarshalText() (ret []byte, err error)

func (RetryStrategy) String

func (strategy RetryStrategy) String() string

func (*RetryStrategy) UnmarshalText

func (strategy *RetryStrategy) UnmarshalText(data []byte) (err error)

UnmarshalText implements encoding.TextUnmarshaler.