queue

Documentation

Index

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

Constants

const (
	ErrJobDuplicated errhelper.ErrString = "job duplicat"
	ErrJobConflict   errhelper.ErrString = "job conflict"
	ErrJobCounteract errhelper.ErrString = "job counteract"
	ErrJobInvalid    errhelper.ErrString = "job invalid"
)

Errors for JobQueue

const (
	ErrNotStarted    errhelper.ErrString = "not started"
	ErrStopped       errhelper.ErrString = "stopped"
	ErrKeyNotAllowed errhelper.ErrString = "key not allowed"
)

Errors for timeout queue

Types

type Job

type Job[K comparable] struct {
	Action JobAction
	Key    K
}

Job item to record action and related resource object

func (Job[K]) String

func (w Job[K]) 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[K comparable] 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[K comparable]() *JobQueue[K]

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[K]) Acquire

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

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

func (*JobQueue[K]) Find

func (q *JobQueue[K]) Find(key K) (Job[K], bool)

Find the scheduled job according to its key

func (*JobQueue[K]) Offer

func (q *JobQueue[K]) Offer(w Job[K]) error

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

func (*JobQueue[K]) Pause

func (q *JobQueue[K]) Pause()

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

func (*JobQueue[K]) Remains

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

Remains shows what job we are still meant to do

func (*JobQueue[K]) Remove

func (q *JobQueue[K]) Remove(w Job[K]) bool

func (*JobQueue[K]) Resume

func (q *JobQueue[K]) Resume()

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

type SeqDataHandleFunc

type SeqDataHandleFunc[T any] func(seq uint64, d T)

type SeqQueue

type SeqQueue[T any] struct {
	// contains filtered or unexported fields
}

SeqQueue is the sequence queue for unordered data

func NewSeqQueue

func NewSeqQueue[T any](handleData SeqDataHandleFunc[T]) *SeqQueue[T]

NewSeqQueue returns a empty SeqQueue

func (*SeqQueue[T]) Offer

func (q *SeqQueue[T]) Offer(seq uint64, data T) (complete bool)

Offer an unordered data with its sequence

func (*SeqQueue[T]) Reset

func (q *SeqQueue[T]) Reset()

Reset the SeqQueue for new sequential data

func (*SeqQueue[T]) SetMaxSeq

func (q *SeqQueue[T]) SetMaxSeq(maxSeq uint64) (complete bool)

SetMaxSeq set when should this queue stop enqueuing data

type TimeoutData

type TimeoutData[K comparable, V any] struct {
	Key  K
	Data V
	// contains filtered or unexported fields
}

TimeoutData is the data set used internally

type TimeoutQueue

type TimeoutQueue[K comparable, V any] 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[K comparable, V any]() *TimeoutQueue[K, V]

NewTimeoutQueue returns an idle TimeoutQueue

func (*TimeoutQueue[K, V]) Allow

func (q *TimeoutQueue[K, V]) Allow(key K)

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

func (*TimeoutQueue[K, V]) Clear

func (q *TimeoutQueue[K, V]) Clear()

Clear out all timeout key-value pairs

func (*TimeoutQueue[K, V]) Find

func (q *TimeoutQueue[K, V]) Find(key K) (_ V, _ bool)

Find timeout key-value pair according to the key

func (*TimeoutQueue[K, V]) Forbid

func (q *TimeoutQueue[K, V]) Forbid(key K)

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

func (*TimeoutQueue[K, V]) Len

func (q *TimeoutQueue[K, V]) Len() int

Len is used internally for timeout data sort

func (*TimeoutQueue[K, V]) Less

func (q *TimeoutQueue[K, V]) Less(i, j int) bool

Less is used internally for timeout data sort

func (*TimeoutQueue[K, V]) OfferWithDelay

func (q *TimeoutQueue[K, V]) OfferWithDelay(key K, val V, 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[K, V]) OfferWithTime

func (q *TimeoutQueue[K, V]) OfferWithTime(key K, val V, 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[K, V]) Remains

func (q *TimeoutQueue[K, V]) Remains() []TimeoutData[K, V]

Remains shows key-value pairs not timed out

func (*TimeoutQueue[K, V]) Remove

func (q *TimeoutQueue[K, V]) Remove(key K) (ret V, _ bool)

Remove a timeout object from the queue according to the key

func (*TimeoutQueue[K, V]) Start

func (q *TimeoutQueue[K, V]) Start(stop <-chan struct{})

Start routine to generate timeout data

func (*TimeoutQueue[K, V]) Swap

func (q *TimeoutQueue[K, V]) Swap(i, j int)

Swap is used internally for timeout data sort

func (*TimeoutQueue[K, V]) TakeCh

func (q *TimeoutQueue[K, V]) TakeCh() <-chan *TimeoutData[K, V]

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