jobx

Documentation

Index

• Variables
• type DelayQueue
  • func NewDelayQueue(ctx context.Context, dsn string, rdb redis.Cmdable, opts ...Option) (*DelayQueue, error)
  • func (q *DelayQueue) Delete(ctx context.Context, id string) error
  • func (q *DelayQueue) DeleteBatch(ctx context.Context, ids []string) error
  • func (q *DelayQueue) Enqueue(ctx context.Context, job Job) (string, error)
  • func (q *DelayQueue) EnqueueBatch(ctx context.Context, jobs []Job) ([]string, error)
  • func (q *DelayQueue) Shutdown() error
  • func (q *DelayQueue) Start(ctx context.Context, handler Handler) error
• type Handler
• type IDGenerator
• type Job
• type Option
  • func WithBatchSize(n int) Option
  • func WithChunkInterval(d time.Duration) Option
  • func WithConcurrency(n int) Option
  • func WithHotWindow(d time.Duration) Option
  • func WithIDGenerator(fn IDGenerator) Option
  • func WithLeaseTimeout(d time.Duration) Option
  • func WithPollingInterval(d time.Duration) Option
  • func WithPromoteInterval(d time.Duration) Option
  • func WithPromoteRefreshInterval(d time.Duration) Option
  • func WithQueue(queue string) Option
  • func WithQueueDepth(n int) Option
  • func WithRetention(d time.Duration) Option
  • func WithSchema(schema string) Option
  • func WithShardNum(n int) Option
  • func WithSnowflakeDSN(dsn string) Option
  • func WithSnowflakeEpoch(epoch int64) Option
  • func WithSnowflakeTable(table string) Option
  • func WithTable(table string) Option

Variables

var (
	ErrNilPostgres     = errors.New("jobx: postgres pool is nil")
	ErrNilRedis        = errors.New("jobx: redis client is nil")
	ErrInvalidJob      = errors.New("jobx: invalid job")
	ErrInvalidOption   = errors.New("jobx: invalid option")
	ErrQueueClosed     = errors.New("jobx: delay queue is closed")
	ErrAlreadyStarted  = errors.New("jobx: delay queue already started")
	ErrMissingHandler  = errors.New("jobx: handler is nil")
	ErrMissingIDSource = errors.New("jobx: id generator returned empty id")
)

Types

type DelayQueue

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

func NewDelayQueue

func NewDelayQueue(ctx context.Context, dsn string, rdb redis.Cmdable, opts ...Option) (*DelayQueue, error)

func (*DelayQueue) Delete

func (q *DelayQueue) Delete(ctx context.Context, id string) error

func (*DelayQueue) DeleteBatch

func (q *DelayQueue) DeleteBatch(ctx context.Context, ids []string) error

func (*DelayQueue) Enqueue

func (q *DelayQueue) Enqueue(ctx context.Context, job Job) (string, error)

func (*DelayQueue) EnqueueBatch

func (q *DelayQueue) EnqueueBatch(ctx context.Context, jobs []Job) ([]string, error)

func (*DelayQueue) Shutdown

func (q *DelayQueue) Shutdown() error

func (*DelayQueue) Start

func (q *DelayQueue) Start(ctx context.Context, handler Handler) error

type Handler

type Handler func(context.Context, []Job) []string

type IDGenerator

type IDGenerator func() string

type Job

type Job struct {
	ID      string
	RunAt   time.Time
	Payload []byte
}

type Option

type Option func(*options)

func WithBatchSize

func WithBatchSize(n int) Option

func WithChunkInterval

func WithChunkInterval(d time.Duration) Option

func WithConcurrency

func WithConcurrency(n int) Option

func WithHotWindow

func WithHotWindow(d time.Duration) Option

func WithIDGenerator

func WithIDGenerator(fn IDGenerator) Option

func WithLeaseTimeout

func WithLeaseTimeout(d time.Duration) Option

func WithPollingInterval

func WithPollingInterval(d time.Duration) Option

func WithPromoteInterval

func WithPromoteInterval(d time.Duration) Option

func WithPromoteRefreshInterval

func WithPromoteRefreshInterval(d time.Duration) Option

func WithQueue

func WithQueue(queue string) Option

func WithQueueDepth

func WithQueueDepth(n int) Option

func WithRetention

func WithRetention(d time.Duration) Option

func WithSchema

func WithSchema(schema string) Option

func WithShardNum

func WithShardNum(n int) Option

func WithSnowflakeDSN

func WithSnowflakeDSN(dsn string) Option

func WithSnowflakeEpoch

func WithSnowflakeEpoch(epoch int64) Option

func WithSnowflakeTable

func WithSnowflakeTable(table string) Option

func WithTable

func WithTable(table string) Option