forked from contribsys/faktory
/
types.go
93 lines (78 loc) · 2.05 KB
/
types.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
package storage
import (
"context"
"fmt"
)
type BackupInfo struct {
Id int64
FileCount int32
Size int64
Timestamp int64
}
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
}
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 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
}
func Open(dbtype string, path string) (Store, error) {
if dbtype == "rocksdb" {
return OpenRocks(path)
} else {
return nil, fmt.Errorf("Invalid dbtype: %s", dbtype)
}
}