engine

Documentation

Index

• Constants
• Variables
• func ExistsPool(pool string) bool
• func GetPools() []string
• func GetPoolsByKind(kind string) []string
• func Register(kind, pool string, e Engine)
• func Shutdown()
• type CreateJobReq
• type Engine
  • func GetEngine(pool string) Engine
  • func GetEngineByKind(kind, pool string) Engine
• type Job
  • func NewJob(namespace, queue string, body []byte, ttl, delay uint32, tries uint16, ...) Job
  • func NewJobFromReq(req *CreateJobReq) Job
  • func NewJobWithID(namespace, queue string, body []byte, ttl uint32, tries uint16, jobID string, ...) Job

Constants

const (
	KindRedis     = "redis"
	KindRedisV2   = "redis_v2"
	KindMigration = "migration"
)

Variables

var (
	ErrNotFound   = errors.New("job not found")
	ErrEmptyQueue = errors.New("the queue is empty")
	ErrWrongQueue = errors.New("wrong queue for the job")
)

Functions

func ExistsPool

func ExistsPool(pool string) bool

func GetPools

func GetPools() []string

func GetPoolsByKind

func GetPoolsByKind(kind string) []string

func Register

func Register(kind, pool string, e Engine)

func Shutdown

func Shutdown()

Types

type CreateJobReq

type CreateJobReq struct {
	Namespace  string
	Queue      string
	ID         string
	Body       []byte
	TTL        uint32
	Delay      uint32
	Tries      uint16
	Attributes map[string]string
}

type Engine

type Engine interface {
	Publish(job Job) (jobID string, err error)
	Consume(namespace string, queues []string, ttrSecond, timeoutSecond uint32) (job Job, err error)
	BatchConsume(namespace string, queues []string, count, ttrSecond, timeoutSecond uint32) (jobs []Job, err error)
	Delete(namespace, queue, jobID string) error
	Peek(namespace, queue, optionalJobID string) (job Job, err error)
	Size(namespace, queue string) (size int64, err error)
	Destroy(namespace, queue string) (count int64, err error)

	// Dead letter
	PeekDeadLetter(namespace, queue string) (size int64, jobID string, err error)
	DeleteDeadLetter(namespace, queue string, limit int64) (count int64, err error)
	RespawnDeadLetter(namespace, queue string, limit, ttlSecond int64) (count int64, err error)
	SizeOfDeadLetter(namespace, queue string) (size int64, err error)

	Shutdown()

	DumpInfo(output io.Writer) error
}

func GetEngine

func GetEngine(pool string) Engine

func GetEngineByKind

func GetEngineByKind(kind, pool string) Engine

type Job

type Job interface {
	Namespace() string
	Queue() string
	ID() string
	Body() []byte
	TTL() uint32
	Delay() uint32
	Tries() uint16
	ElapsedMS() int64
	Attributes() map[string]string

	encoding.BinaryMarshaler
	encoding.BinaryUnmarshaler
	encoding.TextMarshaler
}

func NewJob

func NewJob(namespace, queue string, body []byte, ttl, delay uint32, tries uint16, jobID string) Job

NOTE: there is a trick in this factory, the delay is embedded in the jobID. By doing this we can delete the job that's located in hourly AOF, by placing a tombstone record in that AOF.

func NewJobFromReq

func NewJobFromReq(req *CreateJobReq) Job

NewJobFromReq creates a new job with its body and attributes being marshalled

func NewJobWithID

func NewJobWithID(namespace, queue string, body []byte, ttl uint32, tries uint16, jobID string, attrs map[string]string) Job