manager

Documentation

Index

• Constants
• Variables
• func Halt(msg string) error
• type Context
• type Ctx
  • func (c Ctx) Job() *client.Job
  • func (c Ctx) Manager() Manager
• type FailPayload
• type Manager
  • func NewManager(s storage.Store) Manager
• type MiddlewareChain
• type MiddlewareFunc
• type Reservation

Constants

const (
	// Jobs will be reserved for 30 minutes by default.
	// You can customize this per-job with the reserve_for attribute
	// in the job payload.
	DefaultTimeout = 30 * 60

	// Save dead jobs for 180 days, after that they will be purged
	DeadTTL = 180 * 24 * time.Hour
)

Variables

var (
	JobReservationExpired = &FailPayload{
		ErrorType:    "ReservationExpired",
		ErrorMessage: "Faktory job reservation expired",
	}
)

Functions

func Halt

func Halt(msg string) error

Types

type Context

type Context interface {
	context.Context

	Job() *client.Job
	Manager() Manager
}

type Ctx

type Ctx struct {
	context.Context
	// contains filtered or unexported fields
}

func (Ctx) Job

func (c Ctx) Job() *client.Job

func (Ctx) Manager

func (c Ctx) Manager() Manager

type FailPayload

type FailPayload struct {
	Jid          string   `json:"jid"`
	ErrorMessage string   `json:"message"`
	ErrorType    string   `json:"errtype"`
	Backtrace    []string `json:"backtrace"`
}

type Manager

type Manager interface {
	Push(job *client.Job) error

	// Dispatch operations:
	//
	//  - Basic dequeue
	//    - Connection sends FETCH q1, q2
	// 	 - Job moved from Queue into Working
	//  - Scheduled
	//  	 - Job Pushed into Queue
	// 	 - Job moved from Queue into Working
	//  - Failure
	//    - Job Pushed into Retries
	//  - Push
	//    - Job Pushed into Queue
	//  - Ack
	//    - Job removed from Working
	//
	// How are jobs passed to waiting workers?
	//
	// Socket sends "FETCH q1, q2, q3"
	// Connection pops each queue:
	//   store.GetQueue("q1").Pop()
	// and returns if it gets any non-nil data.
	//
	// If all nil, the connection registers itself, blocking for a job.
	Fetch(ctx context.Context, wid string, queues ...string) (*client.Job, error)

	Acknowledge(jid string) (*client.Job, error)

	Fail(fail *FailPayload) error

	WorkingCount() int

	ReapExpiredJobs(timestamp string) (int, error)

	// Purge deletes all dead jobs
	Purge() (int64, error)

	// EnqueueScheduledJobs enqueues scheduled jobs
	EnqueueScheduledJobs() (int64, error)

	// RetryJobs enqueues failed jobs
	RetryJobs() (int64, error)

	BusyCount(wid string) int

	AddMiddleware(fntype string, fn MiddlewareFunc)

	KV() storage.KV
	Redis() *redis.Client
}

func NewManager

func NewManager(s storage.Store) Manager

type MiddlewareChain

type MiddlewareChain []MiddlewareFunc

type MiddlewareFunc

type MiddlewareFunc func(next func() error, ctx Context) error

type Reservation

type Reservation struct {
	Job    *client.Job `json:"job"`
	Since  string      `json:"reserved_at"`
	Expiry string      `json:"expires_at"`
	Wid    string      `json:"wid"`
	// contains filtered or unexported fields
}