asmt/db/leveldb.go
db/leveldb.goRaw
1/**
2 * @file
3 * @copyright defined in aergo/LICENSE.txt
4 */
5
6package db
7
8import (
9 "bytes"
10 "fmt"
11 "path/filepath"
12
13 "github.com/syndtr/goleveldb/leveldb"
14 "github.com/syndtr/goleveldb/leveldb/errors"
15 "github.com/syndtr/goleveldb/leveldb/iterator"
16 "github.com/syndtr/goleveldb/leveldb/opt"
17)
18
19// This function is always called first
20func init() {
21 dbConstructor := func(dir string) (DB, error) {
22 return newLevelDB(dir)
23 }
24 registorDBConstructor(LevelImpl, dbConstructor)
25}
26
27func newLevelDB(dir string) (DB, error) {
28 dbPath := filepath.Join(dir, "data.db")
29
30 db, err := leveldb.OpenFile(dbPath, nil)
31 if err != nil {
32 return nil, err
33 }
34
35 database := &levelDB{
36 db: db,
37 }
38 return database, nil
39}
40
41//=========================================================
42// DB Implementation
43//=========================================================
44
45// Enforce database and transaction implements interfaces
46var _ DB = (*levelDB)(nil)
47
48type levelDB struct {
49 db *leveldb.DB
50}
51
52func (db *levelDB) Type() string {
53 return "leveldb"
54}
55
56func (db *levelDB) Set(key, value []byte) {
57 key = convNilToBytes(key)
58 value = convNilToBytes(value)
59
60 err := db.db.Put(key, value, &opt.WriteOptions{Sync: true})
61
62 if err != nil {
63 panic(fmt.Sprintf("Database Error: %v", err))
64 }
65}
66
67func (db *levelDB) Delete(key []byte) {
68 key = convNilToBytes(key)
69
70 err := db.db.Delete(key, &opt.WriteOptions{Sync: true})
71
72 if err != nil {
73 panic(fmt.Sprintf("Database Error: %v", err))
74 }
75}
76
77func (db *levelDB) Get(key []byte) []byte {
78 key = convNilToBytes(key)
79 res, err := db.db.Get(key, nil)
80 if err != nil {
81 if err == errors.ErrNotFound {
82 return []byte{}
83 }
84 panic(fmt.Sprintf("Database Error: %v", err))
85 }
86 return res
87}
88
89func (db *levelDB) Exist(key []byte) bool {
90 res, _ := db.db.Has(key, nil)
91 return res
92}
93
94func (db *levelDB) Close() {
95 db.db.Close()
96}
97
98func (db *levelDB) NewTx() Transaction {
99 batch := new(leveldb.Batch)
100
101 return &levelTransaction{db, batch, false, false}
102}
103
104func (db *levelDB) NewBulk() Bulk {
105 batch := new(leveldb.Batch)
106
107 return &levelBulk{db, batch, false, false}
108}
109
110//=========================================================
111// Transaction Implementation
112//=========================================================
113
114type levelTransaction struct {
115 db *levelDB
116 tx *leveldb.Batch
117 isDiscard bool
118 isCommit bool
119}
120
121/*
122func (transaction *levelTransaction) Get(key []byte) []byte {
123 panic(fmt.Sprintf("DO not support"))
124}
125*/
126
127func (transaction *levelTransaction) Set(key, value []byte) {
128 transaction.tx.Put(key, value)
129}
130
131func (transaction *levelTransaction) Delete(key []byte) {
132 transaction.tx.Delete(key)
133}
134
135func (transaction *levelTransaction) Commit() {
136 if transaction.isDiscard {
137 panic("Commit after dicard tx is not allowed")
138 } else if transaction.isCommit {
139 panic("Commit occures two times")
140 }
141 err := transaction.db.db.Write(transaction.tx, &opt.WriteOptions{Sync: true})
142 if err != nil {
143 panic(fmt.Sprintf("Database Error: %v", err))
144 }
145 transaction.isCommit = true
146}
147
148func (transaction *levelTransaction) Discard() {
149 transaction.isDiscard = true
150}
151
152//=========================================================
153// Bulk Implementation
154//=========================================================
155
156type levelBulk struct {
157 db *levelDB
158 tx *leveldb.Batch
159 isDiscard bool
160 isCommit bool
161}
162
163func (bulk *levelBulk) Set(key, value []byte) {
164 bulk.tx.Put(key, value)
165}
166
167func (bulk *levelBulk) Delete(key []byte) {
168 bulk.tx.Delete(key)
169}
170
171func (bulk *levelBulk) Flush() {
172 // do the same behavior that a transaction commit does
173 // db.write internally will handle large transaction
174 if bulk.isDiscard {
175 panic("Commit after dicard tx is not allowed")
176 } else if bulk.isCommit {
177 panic("Commit occures two times")
178 }
179
180 err := bulk.db.db.Write(bulk.tx, &opt.WriteOptions{Sync: true})
181 if err != nil {
182 panic(fmt.Sprintf("Database Error: %v", err))
183 }
184 bulk.isCommit = true
185}
186
187func (bulk *levelBulk) DiscardLast() {
188 bulk.isDiscard = true
189}
190
191//=========================================================
192// Iterator Implementation
193//=========================================================
194
195type levelIterator struct {
196 start []byte
197 end []byte
198 reverse bool
199 iter iterator.Iterator
200 isInvalid bool
201}
202
203func (db *levelDB) Iterator(start, end []byte) Iterator {
204 var reverse bool
205
206 // if end is bigger then start, then reverse order
207 if bytes.Compare(start, end) == 1 {
208 reverse = true
209 } else {
210 reverse = false
211 }
212
213 iter := db.db.NewIterator(nil, nil)
214
215 if reverse {
216 if start == nil {
217 iter.Last()
218 } else {
219 valid := iter.Seek(start)
220 if valid {
221 soakey := iter.Key()
222 if bytes.Compare(start, soakey) < 0 {
223 iter.Prev()
224 }
225 } else {
226 iter.Last()
227 }
228 }
229 } else {
230 if start == nil {
231 iter.First()
232 } else {
233 iter.Seek(start)
234 }
235 }
236 return &levelIterator{
237 iter: iter,
238 start: start,
239 end: end,
240 reverse: reverse,
241 isInvalid: false,
242 }
243}
244
245func (iter *levelIterator) Next() {
246 if iter.Valid() {
247 if iter.reverse {
248 iter.iter.Prev()
249 } else {
250 iter.iter.Next()
251 }
252 } else {
253 panic("Iterator is Invalid")
254 }
255}
256
257func (iter *levelIterator) Valid() bool {
258
259 // Once invalid, forever invalid.
260 if iter.isInvalid {
261 return false
262 }
263
264 // Panic on DB error. No way to recover.
265 if err := iter.iter.Error(); err != nil {
266 panic(err)
267 }
268
269 // If source is invalid, invalid.
270 if !iter.iter.Valid() {
271 iter.isInvalid = true
272 return false
273 }
274
275 // If key is end or past it, invalid.
276 var end = iter.end
277 var key = iter.iter.Key()
278
279 if iter.reverse {
280 if end != nil && bytes.Compare(key, end) <= 0 {
281 iter.isInvalid = true
282 return false
283 }
284 } else {
285 if end != nil && bytes.Compare(end, key) <= 0 {
286 iter.isInvalid = true
287 return false
288 }
289 }
290
291 // Valid
292 return true
293}
294
295func (iter *levelIterator) Key() (key []byte) {
296 if !iter.Valid() {
297 panic("Iterator is invalid")
298 } else if err := iter.iter.Error(); err != nil {
299 panic(err)
300 }
301
302 originalKey := iter.iter.Key()
303
304 key = make([]byte, len(originalKey))
305 copy(key, originalKey)
306
307 return key
308}
309
310func (iter *levelIterator) Value() (value []byte) {
311 if !iter.Valid() {
312 panic("Iterator is invalid")
313 } else if err := iter.iter.Error(); err != nil {
314 panic(err)
315 }
316 originalValue := iter.iter.Value()
317
318 value = make([]byte, len(originalValue))
319 copy(value, originalValue)
320
321 return value
322}