queue

Documentation

Index

• type AlreadyExistsError
  • func (ae AlreadyExistsError) Error() string
• type Element
• type Queue
  • func MakeQueue(persist persister.Persister) Queue
• type QueueState
• type VersionError
  • func (ve VersionError) Error() string
• type VersionedQueue
  • func VQ(persist persister.Persister) (vq *VersionedQueue)
  • func (vq *VersionedQueue) Find(id string) (pos int, seq int64, err error)
  • func (vq *VersionedQueue) Get(i int, seq int64) (el Element, nseq int64, err error)
  • func (vq *VersionedQueue) List() (els []Element, seq int64)
  • func (vq *VersionedQueue) Move(i int, npos int, seq int64) (nseq int64, err error)
  • func (vq *VersionedQueue) Persist()
  • func (vq *VersionedQueue) Put(el Element) (pos int, seq int64, err error)
  • func (vq *VersionedQueue) Recover()
  • func (vq *VersionedQueue) Remove(i int, seq int64) (nseq int64, err error)
  • func (vq *VersionedQueue) Size() (size int, seq int64)
  • func (vq *VersionedQueue) Take(i int, seq int64) (el Element, nseq int64, err error)
  • func (vq *VersionedQueue) TakeFront() (el Element, seq int64, err error)

Types

type AlreadyExistsError

type AlreadyExistsError struct {
	Id        string
	Timestamp time.Time
}

func (AlreadyExistsError) Error

func (ae AlreadyExistsError) Error() string

type Element

type Element struct {
	Id       string    `json:"Id"`
	Metadata string    `json:"Metadata"`
	QTime    time.Time `json:"QTime"`
}

type Queue

type Queue interface {
	Put(el Element) (pos int, err error)
	TakeFront() (el Element, err error)
	Take(i int) (el Element, err error)
	Get(i int) (el Element, err error)
	Remove(i int) (err error)
	Move(i int, npos int) (err error)
	Find(id string) (pos int, err error)
	List() (els []Element)
	Size() int

	Persist()
	Recover()
}

func MakeQueue

func MakeQueue(persist persister.Persister) Queue

type QueueState

type QueueState struct {
	Elements []Element `json:"Elements"`
}

type VersionError

type VersionError struct {
	Current   int64 // current sequence number
	Attempted int64
}

func (VersionError) Error

func (ve VersionError) Error() string

type VersionedQueue

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

Versioned queue for concurrent and asynchronous modifications.

All operations return a version number associated with the state of an underlying queue. Modification operations must supply a version number, except for blind writes (e.g., Put and TakeFront). All possible modifications to the backing queue increase the version number.

Version mismatches return a VersionError.

Thread safe.

func VQ

func VQ(persist persister.Persister) (vq *VersionedQueue)

func (*VersionedQueue) Find

func (vq *VersionedQueue) Find(id string) (pos int, seq int64, err error)

func (*VersionedQueue) Get

func (vq *VersionedQueue) Get(i int, seq int64) (el Element, nseq int64, err error)

func (*VersionedQueue) List

func (vq *VersionedQueue) List() (els []Element, seq int64)

func (*VersionedQueue) Move

func (vq *VersionedQueue) Move(i int, npos int, seq int64) (nseq int64, err error)

func (*VersionedQueue) Persist

func (vq *VersionedQueue) Persist()

func (*VersionedQueue) Put

func (vq *VersionedQueue) Put(el Element) (pos int, seq int64, err error)

func (*VersionedQueue) Recover

func (vq *VersionedQueue) Recover()

func (*VersionedQueue) Remove

func (vq *VersionedQueue) Remove(i int, seq int64) (nseq int64, err error)

func (*VersionedQueue) Size

func (vq *VersionedQueue) Size() (size int, seq int64)

func (*VersionedQueue) Take

func (vq *VersionedQueue) Take(i int, seq int64) (el Element, nseq int64, err error)

func (*VersionedQueue) TakeFront

func (vq *VersionedQueue) TakeFront() (el Element, seq int64, err error)