-
Notifications
You must be signed in to change notification settings - Fork 0
/
store.go
125 lines (100 loc) · 2.52 KB
/
store.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
116
117
118
119
120
121
122
123
124
125
// SPDX-FileCopyrightText: 2023 Steffen Vogel <post@steffenvogel.de>
// SPDX-License-Identifier: Apache-2.0
package mqtt
import (
"bytes"
"fmt"
"log"
"path"
"github.com/dgraph-io/badger/v4"
"github.com/eclipse/paho.mqtt.golang/packets"
)
type store struct {
*badger.DB
path string
}
func newStore(clientID, dataDir string) (*store, error) {
return &store{
path: path.Join(dataDir, "mqtt", fmt.Sprintf("%s.db", clientID)),
}, nil
}
func (s *store) Open() {
var err error
opts := badger.DefaultOptions(s.path)
opts.ValueLogFileSize = 1 << 20
if s.DB, err = badger.Open(opts); err != nil {
log.Fatalf("Failed to open database: %s", err)
}
}
func (s *store) Put(key string, msg packets.ControlPacket) {
wr := bytes.NewBuffer(nil)
if err := msg.Write(wr); err != nil {
log.Fatalf("Failed to write: %s", err)
}
if err := s.DB.Update(func(txn *badger.Txn) error {
return txn.Set([]byte(key), wr.Bytes())
}); err != nil {
log.Fatalf("Failed to list all keys: %s", err)
}
}
func (s *store) Get(key string) packets.ControlPacket {
buf := bytes.NewBuffer(nil)
if err := s.DB.View(func(txn *badger.Txn) error {
if it, err := txn.Get([]byte(key)); err == nil {
return it.Value(func(val []byte) error {
_, err := buf.Write(val)
return err
})
} else {
return err
}
}); err != nil {
log.Fatalf("Failed to list all keys: %s", err)
}
pkt, _ := packets.ReadPacket(buf)
return pkt
}
func (s *store) All() []string {
all := []string{}
if err := s.DB.View(func(txn *badger.Txn) error {
opts := badger.DefaultIteratorOptions
opts.PrefetchValues = false
it := txn.NewIterator(opts)
defer it.Close()
for it.Rewind(); it.Valid(); it.Next() {
all = append(all, string(it.Item().Key()))
}
return nil
}); err != nil {
log.Fatalf("Failed to list all keys: %s", err)
}
return all
}
func (s *store) Del(key string) {
if err := s.DB.Update(func(txn *badger.Txn) error {
return txn.Delete([]byte(key))
}); err != nil {
log.Fatalf("Failed to list all keys: %s", err)
}
}
func (s *store) Close() {
if err := s.DB.Close(); err != nil {
log.Fatalf("Failed to close database: %s", err)
}
}
func (s *store) Reset() {
if err := s.DB.Update(func(txn *badger.Txn) error {
opts := badger.DefaultIteratorOptions
opts.PrefetchValues = false
it := txn.NewIterator(opts)
defer it.Close()
for it.Rewind(); it.Valid(); it.Next() {
if err := txn.Delete(it.Item().Key()); err != nil {
return err
}
}
return nil
}); err != nil {
log.Fatalf("failed to reset store: %s", err)
}
}