jobqueue

type BackoffFunc func(attempts int) time.Duration

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

type Job struct {
	ID               string        `json:"id"`        // internal identifier
	Topic            string        `json:"topic"`     // topic to find the correct processor
	State            string        `json:"state"`     // current state
	Args             []interface{} `json:"args"`      // arguments to pass to processor
	Rank             int           `json:"rank"`      // jobs with higher ranks get executed earlier
	Priority         int64         `json:"prio"`      // priority (highest gets executed first)
	Retry            int           `json:"retry"`     // current number of retries
	MaxRetry         int           `json:"maxretry"`  // maximum number of retries
	CorrelationGroup string        `json:"cgroup"`    // external group
	CorrelationID    string        `json:"cid"`       // external identifier
	Created          int64         `json:"created"`   // time when Add was called (in UnixNano)
	Updated          int64         `json:"updated"`   // time when the job was last updated (in UnixNano)
	Started          int64         `json:"started"`   // time when the job was started (in UnixNano)
	Completed        int64         `json:"completed"` // time when job reached either state Succeeded or Failed (in UnixNano)
}

type ListRequest struct {
	Topic            string // filter by topic
	CorrelationGroup string // filter by correlation group
	CorrelationID    string // filter by correlation identifier
	State            string // filter by job state
	Limit            int    // maximum number of jobs to return
	Offset           int    // number of jobs to skip (for pagination)
}

type ListResponse struct {
	Total int    // total number of jobs found, excluding pagination
	Jobs  []*Job // list of jobs
}

type Logger interface {
	Printf(format string, v ...interface{})
}

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

type ManagerOption func(*Manager)

type Processor func(*Job) error

type StartupBehaviour int

type Stats struct {
	Waiting   int `json:"waiting"`   // number of jobs waiting to be executed
	Working   int `json:"working"`   // number of jobs currently being executed
	Succeeded int `json:"succeeded"` // number of successfully completed jobs
	Failed    int `json:"failed"`    // number of failed jobs (even after retries)
}

type StatsRequest struct {
	Topic            string // filter by topic
	CorrelationGroup string // filter by correlation group
}

type Store interface {
	// Start is called when the manager starts up.
	// This is a good time for cleanup. E.g. a persistent store might moved
	// crashed jobs from a previous run into the Failed state.
	Start(StartupBehaviour) error

	// Create adds a job to the store.
	Create(context.Context, *Job) error

	// Delete removes a job from the store.
	Delete(context.Context, *Job) error

	// Update updates a job in the store. This is called frequently as jobs
	// are processed. Update must allow for concurrent updates, e.g. by locking.
	Update(context.Context, *Job) error

	// Next picks the next job to execute.
	//
	// The store should take the job priorities into account when picking the
	// next job. Jobs with higher priorities should be executed first.
	//
	// If no job is ready to be executed, e.g. the job queue is idle, the
	// store must return nil for both the job and the error.
	Next() (*Job, error)

	// Stats returns statistics about the store, e.g. the number of jobs
	// waiting, working, succeeded, and failed. This is run when the manager
	// starts up to get initial stats.
	Stats(context.Context, *StatsRequest) (*Stats, error)

	// Lookup returns the details of a job by its identifier.
	// If the job could not be found, ErrNotFound must be returned.
	Lookup(context.Context, string) (*Job, error)

	// LookupByCorrelationID returns the details of jobs by their correlation identifier.
	// If no such job could be found, an empty array is returned.
	LookupByCorrelationID(context.Context, string) ([]*Job, error)

	// List returns a list of jobs filtered by the ListRequest.
	List(context.Context, *ListRequest) (*ListResponse, error)
}