-
Notifications
You must be signed in to change notification settings - Fork 0
/
mem.go
115 lines (102 loc) · 2.32 KB
/
mem.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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
package reconEngine
import (
"bytes"
"errors"
"strings"
"sync"
)
//Base mem interface, you can implement own realisation
type MemStorage interface {
Get(key string) ([]byte, error)
Set(key string, value []byte)
Del(key string) error
Scan(subStr string, cb func(key string, value []byte) bool)
Sync() error
Len() int
SsTable() SsTableStorage
}
// Memory table
type mem struct {
storage map[string][]byte
ssTable SsTableStorage
}
func (m *mem) Scan(subStr string, cb func(key string, value []byte) bool) {
needNextIteration := true
for key, value := range m.storage {
if strings.Contains(key, subStr) {
if !cb(key, value) {
return
}
}
}
m.ssTable.Range(func(createdAt int64, partitionStorage SsTablePartitionStorage) bool {
partitionStorage.Range(func(key string, value []byte) bool {
if _, ok := m.storage[key]; !ok && strings.Contains(key, subStr) {
m.Set(key, value)
if !cb(key, value) {
needNextIteration = false
}
}
return needNextIteration
})
return needNextIteration
})
}
// Key not found error
var KeyNotFoundErr = errors.New("can't found value by that key")
// Removed key error
var KeyRemovedErr = errors.New("that key was removed")
func (m *mem) Get(key string) ([]byte, error) {
val, ok := m.storage[key]
if ok && bytes.Equal(val, []byte{removed}) {
return nil, KeyNotFoundErr
}
if !ok {
val, err := m.ssTable.Get(key)
if err == nil {
m.storage[key] = val
}
return val, err
}
return val, nil
}
func (m *mem) Set(key string, value []byte) {
m.storage[key] = value
}
func (m *mem) Del(key string) error {
m.storage[key] = []byte{removed}
return nil
}
func (m *mem) Sync() error {
var mx sync.Mutex
mx.Lock()
defer mx.Unlock()
ssp := m.ssTable.CreatePartition()
for k, v := range m.storage {
err := ssp.Set(k, v)
if err != nil {
return err
} else {
delete(m.storage, k)
}
}
return nil
}
func (m *mem) Len() int {
return len(m.storage)
}
func (m *mem) SsTable() SsTableStorage {
return m.ssTable
}
// Mem constructor, create structure realised MemStorage interface
// ssTable argument may be a nil
// dir argument that will be stored data for partitions
func NewMem(ssTable SsTableStorage, dir *string) MemStorage {
if ssTable == nil {
ssTable = NewSsTable(dir)
}
return &mem{
storage: make(map[string][]byte),
ssTable: ssTable,
}
}