queue

Documentation

Index

• Variables
• type Job
  • func (w Job) String() string
• type JobAction
  • func (t JobAction) String() string
• type JobQueue
  • func NewJobQueue() *JobQueue
  • func (q *JobQueue) Acquire() (w Job, shouldAcquireMore bool)
  • func (q *JobQueue) Find(key interface{}) (Job, bool)
  • func (q *JobQueue) Offer(w Job) error
  • func (q *JobQueue) Pause()
  • func (q *JobQueue) Remains() []Job
  • func (q *JobQueue) Remove(w Job) bool
  • func (q *JobQueue) Resume()
• type SeqDataHandleFunc
• type SeqQueue
  • func NewSeqQueue(handleData SeqDataHandleFunc) *SeqQueue
  • func (q *SeqQueue) Offer(seq uint64, data interface{}) (complete bool)
  • func (q *SeqQueue) Reset()
  • func (q *SeqQueue) SetMaxSeq(maxSeq uint64) (complete bool)
• type TimeoutData
• type TimeoutQueue
  • func NewTimeoutQueue() *TimeoutQueue
  • func (q *TimeoutQueue) Allow(key interface{})
  • func (q *TimeoutQueue) Clear()
  • func (q *TimeoutQueue) Find(key interface{}) (interface{}, bool)
  • func (q *TimeoutQueue) Forbid(key interface{})
  • func (q *TimeoutQueue) Len() int
  • func (q *TimeoutQueue) Less(i, j int) bool
  • func (q *TimeoutQueue) OfferWithDelay(key, val interface{}, wait time.Duration) error
  • func (q *TimeoutQueue) OfferWithTime(key, val interface{}, at time.Time) error
  • func (q *TimeoutQueue) Remains() []TimeoutData
  • func (q *TimeoutQueue) Remove(key interface{}) (interface{}, bool)
  • func (q *TimeoutQueue) Start(stop <-chan struct{})
  • func (q *TimeoutQueue) Swap(i, j int)
  • func (q *TimeoutQueue) TakeCh() <-chan *TimeoutData

Variables

var (
	ErrJobDuplicated = errors.New("job duplicat")
	ErrJobConflict   = errors.New("job conflict")
	ErrJobCounteract = errors.New("job counteract")
	ErrJobInvalid    = errors.New("job invalid")
)

Errors for JobQueue

var (
	ErrNotStarted    = errors.New("not started")
	ErrStopped       = errors.New("stopped")
	ErrKeyNotAllowed = errors.New("key not allowed")
)

Errors for timeout queue

Types

type Job

type Job struct {
	Action JobAction
	Key    interface{}
}

Job item to record action and related resource object

func (Job) String

func (w Job) String() string

type JobAction

type JobAction uint8

const (
	// ActionInvalid to do nothing
	ActionInvalid JobAction = iota
	// ActionAdd to add or create some resource
	ActionAdd
	// ActionUpdate to update some resource
	ActionUpdate
	// ActionDelete to delete some resource
	ActionDelete
	// ActionCleanup to eliminate all side effects of the resource
	ActionCleanup
)

func (JobAction) String

func (t JobAction) String() string

type JobQueue

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

JobQueue is the queue data structure designed to reduce redundant job as much as possible

func NewJobQueue

func NewJobQueue() *JobQueue

NewJobQueue will create a stopped new job queue, you can offer job to it, but any acquire will fail until you have called its Resume()

func (*JobQueue) Acquire

func (q *JobQueue) Acquire() (w Job, shouldAcquireMore bool)

Acquire a job item from the job queue if shouldAcquireMore is false, w will be an empty job

func (*JobQueue) Find

func (q *JobQueue) Find(key interface{}) (Job, bool)

Find the scheduled job according to its key

func (*JobQueue) Offer

func (q *JobQueue) Offer(w Job) error

Offer a job item to the job queue if offered job was not added, an error result will return, otherwise nil

func (*JobQueue) Pause

func (q *JobQueue) Pause()

Pause do nothing but mark this job queue is closed, you should not perform acquire actions to the job queue

func (*JobQueue) Remains

func (q *JobQueue) Remains() []Job

Remains shows what job we are still meant to do

func (*JobQueue) Remove

func (q *JobQueue) Remove(w Job) bool

func (*JobQueue) Resume

func (q *JobQueue) Resume()

Resume do nothing but mark you can perform acquire actions to the job queue

type SeqDataHandleFunc

type SeqDataHandleFunc func(seq uint64, d interface{})

type SeqQueue

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

SeqQueue is the sequence queue for unordered data

func NewSeqQueue

func NewSeqQueue(handleData SeqDataHandleFunc) *SeqQueue

NewSeqQueue returns a empty SeqQueue

func (*SeqQueue) Offer

func (q *SeqQueue) Offer(seq uint64, data interface{}) (complete bool)

Offer an unordered data with its sequence

func (*SeqQueue) Reset

func (q *SeqQueue) Reset()

Reset the SeqQueue for new sequential data

func (*SeqQueue) SetMaxSeq

func (q *SeqQueue) SetMaxSeq(maxSeq uint64) (complete bool)

SetMaxSeq set when should this queue stop enqueuing data

type TimeoutData

type TimeoutData struct {
	Key  interface{}
	Data interface{}
	// contains filtered or unexported fields
}

TimeoutData is the data set used internally

type TimeoutQueue

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

TimeoutQueue to arrange timeout events in a single queue, then you can access them in sequence with channel

func NewTimeoutQueue

func NewTimeoutQueue() *TimeoutQueue

NewTimeoutQueue returns an idle TimeoutQueue

func (*TimeoutQueue) Allow

func (q *TimeoutQueue) Allow(key interface{})

Allow allow tasks with key, future tasks with the key can be offered

func (*TimeoutQueue) Clear

func (q *TimeoutQueue) Clear()

Clear out all timeout key-value pairs

func (*TimeoutQueue) Find

func (q *TimeoutQueue) Find(key interface{}) (interface{}, bool)

Find timeout key-value pair according to the key

func (*TimeoutQueue) Forbid

func (q *TimeoutQueue) Forbid(key interface{})

Forbid forbid tasks with key, future tasks with the key cannot be offered

func (*TimeoutQueue) Len

func (q *TimeoutQueue) Len() int

Len is used internally for timeout data sort

func (*TimeoutQueue) Less

func (q *TimeoutQueue) Less(i, j int) bool

Less is used internally for timeout data sort

func (*TimeoutQueue) OfferWithDelay

func (q *TimeoutQueue) OfferWithDelay(key, val interface{}, wait time.Duration) error

OfferWithDelay to enqueue key-value pair, timeout after `wait`, if you would like to call Remove to delete the timeout object, `key` must be unique in this queue

func (*TimeoutQueue) OfferWithTime

func (q *TimeoutQueue) OfferWithTime(key, val interface{}, at time.Time) error

OfferWithTime to enqueue key-value pair with time, timeout at `time`, if you would like to call Remove to delete the timeout object, `key` must be unique in this queue

func (*TimeoutQueue) Remains

func (q *TimeoutQueue) Remains() []TimeoutData

Remains shows key-value pairs not timed out

func (*TimeoutQueue) Remove

func (q *TimeoutQueue) Remove(key interface{}) (interface{}, bool)

Remove a timeout object from the queue according to the key

func (*TimeoutQueue) Start

func (q *TimeoutQueue) Start(stop <-chan struct{})

Start routine to generate timeout data

func (*TimeoutQueue) Swap

func (q *TimeoutQueue) Swap(i, j int)

Swap is used internally for timeout data sort

func (*TimeoutQueue) TakeCh

func (q *TimeoutQueue) TakeCh() <-chan *TimeoutData

TakeCh returns the channel from which you can get key-value pairs timed out one by one