queue

Documentation

Index

• Constants
• type Cfg
• type Opts
• type Queue
  • func New(conn tarantool.Connector, name string) Queue
• type Task
  • func (t *Task) Ack() error
  • func (t *Task) Bury() error
  • func (t *Task) Data() interface{}
  • func (t *Task) DecodeMsgpack(d *msgpack.Decoder) error
  • func (t *Task) Delete() error
  • func (t *Task) Id() uint64
  • func (t *Task) IsBuried() bool
  • func (t *Task) IsDelayed() bool
  • func (t *Task) IsDone() bool
  • func (t *Task) IsReady() bool
  • func (t *Task) IsTaken() bool
  • func (t *Task) Release() error
  • func (t *Task) ReleaseCfg(cfg Opts) error
  • func (t *Task) Status() string

Constants

const (
	READY   = "r"
	TAKEN   = "t"
	DONE    = "-"
	BURIED  = "!"
	DELAYED = "~"
)

const (
	FIFO      queueType = "fifo"
	FIFO_TTL  queueType = "fifottl"
	UTUBE     queueType = "utube"
	UTUBE_TTL queueType = "utubettl"
)

Types

type Cfg

type Cfg struct {
	Temporary   bool // if true, the contents do not persist on disk
	IfNotExists bool // if true, no error will be returned if the tube already exists
	Kind        queueType
	Opts
}

type Opts

type Opts struct {
	Pri   int           // task priorities
	Ttl   time.Duration // task time to live
	Ttr   time.Duration // task time to execute
	Delay time.Duration // delayed execution
	Utube string
}

type Queue

type Queue interface {
	// Exists checks tube for existence
	// Note: it uses Eval, so user needs 'execute universe' privilege
	Exists() (bool, error)
	// Create creates new tube with configuration
	// Note: it uses Eval, so user needs 'execute universe' privilege
	// Note: you'd better not use this function in your application, cause it is
	// administrative task to create or delete queue.
	Create(cfg Cfg) error
	// Drop destroys tube.
	// Note: you'd better not use this function in your application, cause it is
	// administrative task to create or delete queue.
	Drop() error
	// Put creates new task in a tube
	Put(data interface{}) (*Task, error)
	// PutWithOpts creates new task with options different from tube's defaults
	PutWithOpts(data interface{}, cfg Opts) (*Task, error)
	// Take takes 'ready' task from a tube and marks it as 'in progress'
	// Note: if connection has a request Timeout, then 0.9 * connection.Timeout is
	// used as a timeout.
	Take() (*Task, error)
	// TakeWithTimout takes 'ready' task from a tube and marks it as "in progress",
	// or it is timeouted after "timeout" period.
	// Note: if connection has a request Timeout, and conn.Timeout * 0.9 < timeout
	// then timeout = conn.Timeout*0.9
	TakeTimeout(timeout time.Duration) (*Task, error)
	// Take takes 'ready' task from a tube and marks it as 'in progress'
	// Note: if connection has a request Timeout, then 0.9 * connection.Timeout is
	// used as a timeout.
	// Data will be unpacked to result
	TakeTyped(interface{}) (*Task, error)
	// TakeWithTimout takes 'ready' task from a tube and marks it as "in progress",
	// or it is timeouted after "timeout" period.
	// Note: if connection has a request Timeout, and conn.Timeout * 0.9 < timeout
	// then timeout = conn.Timeout*0.9
	// data will be unpacked to result
	TakeTypedTimeout(timeout time.Duration, result interface{}) (*Task, error)
	// Peek returns task by its id.
	Peek(taskId uint64) (*Task, error)
	// Kick reverts effect of Task.Bury() for `count` tasks.
	Kick(count uint64) (uint64, error)
	// Delete the task identified by its id.
	Delete(taskId uint64) error
	// Statistic returns some statistic about queue.
	Statistic() (interface{}, error)
}

Queue is a handle to tarantool queue's tube

func New

func New(conn tarantool.Connector, name string) Queue

New creates a queue handle

type Task

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

Task represents a task from tarantool queue's tube

func (*Task) Ack

func (t *Task) Ack() error

Ack signals about task completion

func (*Task) Bury

func (t *Task) Bury() error

Bury signals that task task cannot be executed in the current circumstances, task becomes "buried" - ie neither completed, nor ready, so it could not be deleted or taken by other worker. To revert "burying" call queue.Kick(numberOfBurried).

func (*Task) Data

func (t *Task) Data() interface{}

Data is a getter for task data

func (*Task) DecodeMsgpack

func (t *Task) DecodeMsgpack(d *msgpack.Decoder) error

func (*Task) Delete

func (t *Task) Delete() error

Delete task from queue

func (*Task) Id

func (t *Task) Id() uint64

Id is a getter for task id

func (*Task) IsBuried

func (t *Task) IsBuried() bool

IsBurred returns if task is buried

func (*Task) IsDelayed

func (t *Task) IsDelayed() bool

IsDelayed returns if task is delayed

func (*Task) IsDone

func (t *Task) IsDone() bool

IsDone returns if task is done

func (*Task) IsReady

func (t *Task) IsReady() bool

IsReady returns if task is ready

func (*Task) IsTaken

func (t *Task) IsTaken() bool

IsTaken returns if task is taken

func (*Task) Release

func (t *Task) Release() error

Release returns task back in the queue without making it complete. In outher words, this worker failed to complete the task, and it, so other worker could try to do that again.

func (*Task) ReleaseCfg

func (t *Task) ReleaseCfg(cfg Opts) error

ReleaseCfg returns task to a queue and changes its configuration.

func (*Task) Status

func (t *Task) Status() string

Status is a getter for task status