backend

Documentation

Overview

Package backend manages storing data.

Index

• Variables
• func Middleware(be Interface) func(next http.Handler) http.Handler
• type Config
• type HandlerProvider
• type Interface
  • func FromMiddleware(ctx context.Context) Interface
• type Invalidator
• type MiddlewareProvider
• type NoopInvalidator
  • func (iv *NoopInvalidator) InvalidateJobs(ctx context.Context) error
• type NoopReaper
  • func (r *NoopReaper) Reap(ctx context.Context, cfg *ReaperConfig) error
• type Reaper
• type ReaperConfig
• type VersionConflictError
  • func NewVersionConflictError(prev, curr *resource.Version, resource, resourceID string) *VersionConflictError
  • func (e *VersionConflictError) Error() string

Variables

var DefaultConfig = Config{
	QueueHistoryLimit: 10,
	ReapAge:           24 * time.Hour * 60,
}

var ErrBlocked = errors.New("backend: queue blocked")

var ErrInvalidResource = errors.New("backend: invalid resource")

ErrInvalidResource should be returned when the resource is missing a field required for storage, like an ID that can't be autogenerated. TODO this should be a struct

var ErrInvalidState = errors.New("backend: invalid state")

var ErrNotAuthorized = errors.New("backend: not authorized")

var ErrNotFound = errors.New("backend: not found")

var (
	JobIncludes = map[string]bool{
		"checkin": true,
		"result":  true,
	}
)

Functions

func Middleware

func Middleware(be Interface) func(next http.Handler) http.Handler

Types

type Config

type Config struct {
	// QueueHistoryLimit is the maximum number of resource versions to store.
	QueueHistoryLimit int           `json:"queue_history_limit" envconfig:"queue_history_limit"`
	ReapAge           time.Duration `json:"reap_age" envconfig:"reap_age"`
	ReapMax           int           `json:"reap_max" envconfig:"reap_max"`
	TestMode          bool          `json:"test_mode" envconfig:"test"`
	Debug             bool

	Logger *logger.Logger
}

type HandlerProvider

type HandlerProvider interface {
	Handler() http.Handler
}

type Interface

type Interface interface {
	Ping(ctx context.Context) error
	// Reset resets the backend. For testing.
	Reset(ctx context.Context) error

	GetQueue(ctx context.Context, job string, opts *resource.GetByIDOpts) (*resource.Queue, error)
	SaveQueue(ctx context.Context, queue *resource.Queue) (*resource.Queue, error)
	ListQueues(ctx context.Context, opts *resource.QueueListParams) (*resource.Queues, error)
	DeleteQueues(ctx context.Context, queues []string) error
	PauseQueues(ctx context.Context, queues []string) error
	UnpauseQueues(ctx context.Context, queues []string) error
	BlockQueues(ctx context.Context, queues []string) error
	UnblockQueues(ctx context.Context, queues []string) error

	EnqueueJobs(ctx context.Context, jobs *resource.Jobs) (*resource.Jobs, error)
	DequeueJobs(ctx context.Context, limit int, opts *resource.JobListParams) (*resource.Jobs, error)
	AckJobs(ctx context.Context, results *resource.Acks) error

	GetJobUniqueArgs(ctx context.Context, keys []string) ([]string, bool, error)
	SetJobUniqueArgs(ctx context.Context, ids, keys []string) error
	DeleteJobUniqueArgs(ctx context.Context, ids, keys []string) error

	GetJobByID(ctx context.Context, id string, opts *resource.GetByIDOpts) (*resource.Job, error)
	ListJobs(ctx context.Context, limit int, opts *resource.JobListParams) (*resource.Jobs, error)

	Stats(ctx context.Context, queue string) (*resource.Stats, error)
}

Interface defines the backend interface. Some required properties:

- UTC

- handle resource versions, conflicts

- safe for concurrent operations

func FromMiddleware

func FromMiddleware(ctx context.Context) Interface

type Invalidator

type Invalidator interface {
	InvalidateJobs(ctx context.Context) error
}

type MiddlewareProvider

type MiddlewareProvider interface {
	Middleware() func(next http.Handler) http.Handler
}

type NoopInvalidator

type NoopInvalidator struct{}

func (*NoopInvalidator) InvalidateJobs

func (iv *NoopInvalidator) InvalidateJobs(ctx context.Context) error

type NoopReaper

type NoopReaper struct{}

func (*NoopReaper) Reap

func (r *NoopReaper) Reap(ctx context.Context, cfg *ReaperConfig) error

type Reaper

type Reaper interface {
	Reap(ctx context.Context, cfg *ReaperConfig) error
}

Reaper adds functionality to clean up old job data.

type ReaperConfig

type ReaperConfig struct {
	Interval    time.Duration `json:"reap_interval"`
	Age         time.Duration `json:"reap_age"`
	Max         int           `json:"reap_max"`
	MaxVersions int           `json:"reap_max_versions"`
}

type VersionConflictError

type VersionConflictError struct {
	Resource   string
	ResourceID string
	Prev       string
	Curr       string
}

func NewVersionConflictError

func NewVersionConflictError(prev, curr *resource.Version, resource, resourceID string) *VersionConflictError

func (*VersionConflictError) Error

func (e *VersionConflictError) Error() string