storage

Documentation

Index

• Variables
• func DefaultOptions() *gorocksdb.Options
• type Backpressure
  • func (bp Backpressure) Error() string
• type BackupInfo
• type Queue
• type QueueWaiter
• type SortedSet
• type Store
  • func Open(dbtype string, path string) (Store, error)
  • func OpenRocks(path string) (Store, error)
• type Transaction

Variables

var (
	/*
	 * The default maximum size of a queue.
	 * Further Pushes will result in an error.
	 *
	 * This is known as "back pressue" and is important to
	 * prevent bugs in one component from taking down the
	 * entire system.
	 */
	DefaultMaxSize = int64(100000)
)

var (
	ONE = make([]byte, 8)
)

Functions

func DefaultOptions

func DefaultOptions() *gorocksdb.Options

Types

type Backpressure

type Backpressure struct {
	QueueName   string
	CurrentSize int64
	MaxSize     int64
}

func (Backpressure) Error

func (bp Backpressure) Error() string

type BackupInfo

type BackupInfo struct {
	Id        int64
	FileCount int32
	Size      int64
	Timestamp int64
}

type Queue

type Queue interface {
	Name() string
	Size() int64
	Push([]byte) error
	Pop() ([]byte, error)
	BPop(context.Context) ([]byte, error)
	Clear() (int64, error)

	/*
	 * Please note that k/vs are NOT safe to use outside of the func.
	 * You must copy the values if you want to stash them for later use.
	 *
	 *	  cpy = make([]byte, len(k))
	 *	  copy(cpy, k)
	 *
	 */
	Each(func(index int, k, v []byte) error) error
	Page(int64, int64, func(index int, k, v []byte) error) error

	Delete(keys [][]byte) error
}

type QueueWaiter

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

type SortedSet

type SortedSet interface {
	Name() string
	Size() int64
	Clear() (int64, error)

	AddElement(timestamp string, jid string, payload []byte) error

	Get(key []byte) ([]byte, error)
	Page(int64, int64, func(index int, key []byte, data []byte) error) error
	Each(func(idx int, key []byte, data []byte) error) error

	Remove(key []byte) error
	RemoveElement(timestamp string, jid string) error
	RemoveBefore(timestamp string) ([][]byte, error)

	/*
		Move the given key from this SortedSet to the given
		SortedSet atomically.  The given func may mutate the payload and
		return a new tstamp.
	*/
	MoveTo(SortedSet, string, string, func([]byte) (string, []byte, error)) error
}

type Store

type Store interface {
	Close() error
	Retries() SortedSet
	Scheduled() SortedSet
	Working() SortedSet
	Dead() SortedSet
	GetQueue(string) (Queue, error)
	EachQueue(func(Queue))
	Stats() map[string]string
	EnqueueAll(SortedSet) error
	EnqueueFrom(SortedSet, []byte) error

	History(days int, fn func(day string, procCnt int64, failCnt int64)) error

	// creates a backup of the current database
	Backup() error
	//Compact() error
	EachBackup(func(bi BackupInfo)) error
	RestoreFromLatest() error

	/*
	 * Clear the database of all job data.
	 * Equivalent to Redis's FLUSHDB
	 */
	Flush() error
}

func Open

func Open(dbtype string, path string) (Store, error)

func OpenRocks

func OpenRocks(path string) (Store, error)

type Transaction

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