tinyq

README

Queue

an implementation of a persistent FIFO queue. with ability to pause/start dequeue.

NOTE: This package is provided "as is" with no guarantee. Use it at your own risk and always test it yourself before using it in a production environment. If you find any issues, please create a new issue.

Install:

go get -u github.com/twiny/tinyq/...

Known Issues:

• all goroutines are asleep - deadlock caused by not closing queue.Dequeue channel.

Documentation

Index

• type Command
• type Message
  • func NewMessage(key string, b interface{}) (Message, error)
  • func (msg Message) Value(v interface{}) error
• type MessageStatus
• type Messages
• type Queue
  • func NewQueue(store Store, autostart bool) *Queue
  • func (q *Queue) Close()
  • func (q *Queue) Dequeue() <-chan Message
  • func (q *Queue) Enqueue(v interface{}) error
  • func (q *Queue) Exec(cmd Command) error
  • func (q *Queue) List(typ MessageStatus, offset, limit uint64) (Messages, error)
  • func (q *Queue) Notify(uuid string, merr error) error
  • func (q *Queue) Remove(typ MessageStatus) error
  • func (q *Queue) Statistic() (Stats, error)
• type State
• type Stats
• type Store

Types

type Command

type Command string

const (
	Start Command = "start"
	Pause Command = "pause"
	Retry Command = "retry"
	Stop  Command = "stop"
)

type Message

type Message struct {
	UUID      string        `json:"uuid"`
	Status    MessageStatus `json:"status"`
	Body      []byte        `json:"body"`
	Detail    string        `json:"detail"`
	Timestamp string        `json:"timestamp"`
}

Message

func NewMessage

func NewMessage(key string, b interface{}) (Message, error)

NewMessage

func (Message) Value

func (msg Message) Value(v interface{}) error

Data

type MessageStatus

type MessageStatus string

MessageStatus

const (
	Pending MessageStatus = "pending"
	Success MessageStatus = "success"
	Failed  MessageStatus = "failed"
)

type Messages

type Messages struct {
	Total  uint64    `json:"total"`
	Offset uint64    `json:"offset"`
	Limit  uint64    `json:"limit"`
	Page   uint64    `json:"page"`
	Items  []Message `json:"items"`
}

Messages

type Queue

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

Queue

func NewQueue

func NewQueue(store Store, autostart bool) *Queue

NewQueue

func (*Queue) Close

func (q *Queue) Close()

Close

func (*Queue) Dequeue

func (q *Queue) Dequeue() <-chan Message

Dequeue

func (*Queue) Enqueue

func (q *Queue) Enqueue(v interface{}) error

Enqueue

func (*Queue) Exec

func (q *Queue) Exec(cmd Command) error

Exec

func (*Queue) List

func (q *Queue) List(typ MessageStatus, offset, limit uint64) (Messages, error)

List

func (*Queue) Notify

func (q *Queue) Notify(uuid string, merr error) error

Notify

func (*Queue) Remove

func (q *Queue) Remove(typ MessageStatus) error

Remove

func (*Queue) Statistic

func (q *Queue) Statistic() (Stats, error)

Statistic

type State

type State string

State

type Stats

type Stats struct {
	IsRunning bool   `json:"is_running"`
	Pending   uint64 `json:"pending"`
	Failed    uint64 `json:"failed"`
}

Stats

type Store

type Store interface {
	Enqueue(msg Message) error
	Dequeue() (Message, error)
	Retry() error
	Notify(uuid string, merr error) error
	List(typ MessageStatus, offset, limit uint64) (Messages, error)
	IsEmpty() bool
	Remove(typ MessageStatus) error
	Statistic() (Stats, error)
	Close()
}

Store