dqdef

Documentation

Index

• Variables
• type BlockDelayQueue
• type DelayQueue
• type FNProcessJob
• type Job
• type JobIdentify
• type JobPool
• type JobPoolManager
• type NotifyDelayQueue
• type ReadyPool
• type ReadyPoolFetcher
• type ReadyPoolManager
• type ReadyPoolNotifier

Variables

var (
	ErrSafeJob    = cuserror.NewWithErrorMsg("isSafeJob")
	ErrNoSafeJob  = cuserror.NewWithErrorMsg("isNoSafeJob")
	ErrTimeout    = cuserror.NewWithErrorMsg("timeout")
	ErrInvalidJob = cuserror.NewWithErrorMsg("invalidJob")
)

Types

type BlockDelayQueue

type BlockDelayQueue interface {
	PushJob(job *Job) error

	// PushSafeJob job到期后会归到完成池，但是仍然会在延迟队列保持TTR的时间，如果TTR时间过后
	// 还没有被调用者主动从延迟队列删除，则会重新调度到完成池.
	PushSafeJob(job *Job) error

	BlockProcessJobOnce(f FNProcessJob, timeout time.Duration, jobIn *Job, topics ...string) (ok bool, err error)

	StopAndWait()
	Wait()

	GetDelayQueue() DelayQueue
}

type DelayQueue

type DelayQueue interface {
	GetReadyPool() ReadyPool
	GetJobPool() JobPool

	JobPush(job *Job) error
	JobDone(jobID string)

	StopAndWait()
	Wait()
}

type FNProcessJob

type FNProcessJob func(job *Job) (newJob *Job, err error)

FNProcessJob 返回值err非空，则安全Job会被再次调度，非安全Job终止继续操作 返回值newJob非空，则新的任务会加入延迟队列 注意: 会调用不能添加和当前job相同ID的新任务到延迟队列

type Job

type Job struct {
	Topic string        `json:"topic" msgpack:"1"`
	ID    string        `json:"id" msgpack:"2"`    // job唯一标识ID
	Delay time.Time     `json:"delay" msgpack:"3"` // 延迟时间
	TTR   time.Duration `json:"ttr" msgpack:"4"`   // 最大任务执行时间, <0代表不设置
	Body  string        `json:"body" msgpack:"5"`
	BodyO interface{}   `json:"body_o" msgpack:"6"`
}

type JobIdentify

type JobIdentify struct {
	Topic string
	ID    string
}

type JobPool

type JobPool interface {
	SaveJob(ctx context.Context, job *Job, afterHook func() error) (err error)
	GetJob(ctx context.Context, jobID string, jobIn *Job) (job *Job, err error)
	// RemoveJobByDQ FIXME hide it.
	RemoveJobByDQ(ctx context.Context, jobID string) (err error)
	RemoveJob(ctx context.Context, jobID string) (err error)
}

type JobPoolManager

type JobPoolManager interface {
	JobPool
	SetDelayQueueName(name string)
}

type NotifyDelayQueue

type NotifyDelayQueue interface {
	PushJob(job *Job) error
	ReadyJobChannel() <-chan *Job

	StopAndWait()
	Wait()
}

type ReadyPool

type ReadyPool interface {
	NewReadyJob(topic, jobID string) (err error)
}

type ReadyPoolFetcher

type ReadyPoolFetcher interface {
	GetReadyJob(timeout time.Duration, topics ...string) (jid *JobIdentify, err error)
}

type ReadyPoolManager

type ReadyPoolManager interface {
	ReadyPool
	SetDelayQueueName(name string)
}

type ReadyPoolNotifier

type ReadyPoolNotifier interface {
	JobChan() <-chan *JobIdentify
}