-
Notifications
You must be signed in to change notification settings - Fork 47
/
Copy pathput.go
295 lines (238 loc) · 7.51 KB
/
put.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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
// Copyright 2016 Tim Shannon. All rights reserved.
// Use of this source code is governed by the MIT license
// that can be found in the LICENSE file.
package bolthold
import (
"errors"
"reflect"
bolt "go.etcd.io/bbolt"
)
// ErrKeyExists is the error returned when data is being Inserted for a Key that already exists
var ErrKeyExists = errors.New("This Key already exists in this bolthold for this type")
// ErrUniqueExists is the error thrown when data is being inserted for a unique constraint value that already exists
var ErrUniqueExists = errors.New("This value cannot be written due to the unique constraint on the field")
// sequence tells bolthold to insert the key as the next sequence in the bucket
type sequence struct{}
// NextSequence is used to create a sequential key for inserts
// Inserts a uint64 as the key
// store.Insert(bolthold.NextSequence(), data)
func NextSequence() interface{} {
return sequence{}
}
// Insert inserts the passed in data into the the bolthold
//
// If the the key already exists in the bolthold, then an ErrKeyExists is returned
// If the data struct has a field tagged as `boltholdKey` and it is the same type
// as the Insert key, AND the data struct is passed by reference, AND the key field
// is currently set to the zero-value for that type, then that field will be set to
// the value of the insert key.
//
// To use this with bolthold.NextSequence() use a type of `uint64` for the key field.
func (s *Store) Insert(key, data interface{}) error {
return s.Bolt().Update(func(tx *bolt.Tx) error {
return s.insert(tx, key, data)
})
}
// TxInsert is the same as Insert except it allows you specify your own transaction
func (s *Store) TxInsert(tx *bolt.Tx, key, data interface{}) error {
if !tx.Writable() {
return bolt.ErrTxNotWritable
}
return s.insert(tx, key, data)
}
// InsertIntoBucket is the same as Insert except it allows you specify your own parent bucket
func (s *Store) InsertIntoBucket(parent *bolt.Bucket, key, data interface{}) error {
if !parent.Tx().Writable() {
return bolt.ErrTxNotWritable
}
return s.insert(parent, key, data)
}
func (s *Store) insert(source BucketSource, key, data interface{}) error {
storer := s.newStorer(data)
b, err := source.CreateBucketIfNotExists([]byte(storer.Type()))
if err != nil {
return err
}
if _, ok := key.(sequence); ok {
key, err = b.NextSequence()
if err != nil {
return err
}
}
gk, err := s.encode(key)
if err != nil {
return err
}
if b.Get(gk) != nil {
return ErrKeyExists
}
value, err := s.encode(data)
if err != nil {
return err
}
// insert data
err = b.Put(gk, value)
if err != nil {
return err
}
// insert any indexes
err = s.addIndexes(storer, source, gk, data)
if err != nil {
return err
}
dataVal := reflect.Indirect(reflect.ValueOf(data))
if !dataVal.CanSet() {
return nil
}
dataType := dataVal.Type()
for i := 0; i < dataType.NumField(); i++ {
tf := dataType.Field(i)
// XXX: should we require standard tag format so we can use StructTag.Lookup()?
// XXX: should we use strings.Contains(string(tf.Tag), BoltholdKeyTag) so we don't require proper tags?
if _, ok := tf.Tag.Lookup(BoltholdKeyTag); ok {
fieldValue := dataVal.Field(i)
keyValue := reflect.ValueOf(key)
if keyValue.Type() != tf.Type {
break
}
if !fieldValue.CanSet() {
break
}
if !reflect.DeepEqual(fieldValue.Interface(), reflect.Zero(tf.Type).Interface()) {
break
}
fieldValue.Set(keyValue)
break
}
}
return nil
}
// Update updates an existing record in the bolthold
// if the Key doesn't already exist in the store, then it fails with ErrNotFound
func (s *Store) Update(key interface{}, data interface{}) error {
return s.Bolt().Update(func(tx *bolt.Tx) error {
return s.update(tx, key, data)
})
}
// TxUpdate is the same as Update except it allows you to specify your own transaction
func (s *Store) TxUpdate(tx *bolt.Tx, key interface{}, data interface{}) error {
if !tx.Writable() {
return bolt.ErrTxNotWritable
}
return s.update(tx, key, data)
}
// UpdateBucket allows you to run an update against any parent bucket
func (s *Store) UpdateBucket(parent *bolt.Bucket, key interface{}, data interface{}) error {
if !parent.Tx().Writable() {
return bolt.ErrTxNotWritable
}
return s.update(parent, key, data)
}
func (s *Store) update(source BucketSource, key interface{}, data interface{}) error {
storer := s.newStorer(data)
gk, err := s.encode(key)
if err != nil {
return err
}
b, err := source.CreateBucketIfNotExists([]byte(storer.Type()))
if err != nil {
return err
}
existing := b.Get(gk)
if existing == nil {
return ErrNotFound
}
// delete any existing indexes
existingVal := newElemType(data)
err = s.decode(existing, existingVal)
if err != nil {
return err
}
err = s.deleteIndexes(storer, source, gk, existingVal)
if err != nil {
return err
}
value, err := s.encode(data)
if err != nil {
return err
}
// put data
err = b.Put(gk, value)
if err != nil {
return err
}
// insert any new indexes
return s.addIndexes(storer, source, gk, data)
}
// Upsert inserts the record into the bolthold if it doesn't exist. If it does already exist, then it updates
// the existing record
func (s *Store) Upsert(key interface{}, data interface{}) error {
return s.Bolt().Update(func(tx *bolt.Tx) error {
return s.upsert(tx, key, data)
})
}
// TxUpsert is the same as Upsert except it allows you to specify your own transaction
func (s *Store) TxUpsert(tx *bolt.Tx, key interface{}, data interface{}) error {
if !tx.Writable() {
return bolt.ErrTxNotWritable
}
return s.upsert(tx, key, data)
}
// UpsertBucket allows you to run an upsert against any bucket parent
func (s *Store) UpsertBucket(parent *bolt.Bucket, key interface{}, data interface{}) error {
if !parent.Tx().Writable() {
return bolt.ErrTxNotWritable
}
return s.upsert(parent, key, data)
}
func (s *Store) upsert(source BucketSource, key interface{}, data interface{}) error {
storer := s.newStorer(data)
gk, err := s.encode(key)
if err != nil {
return err
}
b, err := source.CreateBucketIfNotExists([]byte(storer.Type()))
if err != nil {
return err
}
existing := b.Get(gk)
if existing != nil {
existingVal := newElemType(data)
err = s.decode(existing, existingVal)
if err != nil {
return err
}
err = s.deleteIndexes(storer, source, gk, existingVal)
if err != nil {
return err
}
}
value, err := s.encode(data)
if err != nil {
return err
}
// put data
err = b.Put(gk, value)
if err != nil {
return err
}
// insert any new indexes
return s.addIndexes(storer, source, gk, data)
}
// UpdateMatching runs the update function for every record that match the passed in query
// Note that the type of record in the update func always has to be a pointer
func (s *Store) UpdateMatching(dataType interface{}, query *Query, update func(record interface{}) error) error {
return s.Bolt().Update(func(tx *bolt.Tx) error {
return s.updateQuery(tx, dataType, query, update)
})
}
// TxUpdateMatching does the same as UpdateMatching, but allows you to specify your own transaction
func (s *Store) TxUpdateMatching(tx *bolt.Tx, dataType interface{}, query *Query,
update func(record interface{}) error) error {
return s.updateQuery(tx, dataType, query, update)
}
// UpdateMatchingInBucket does the same as UpdateMatching, but allows you to specify your own parent bucket
func (s *Store) UpdateMatchingInBucket(parent *bolt.Bucket, dataType interface{}, query *Query,
update func(record interface{}) error) error {
return s.updateQuery(parent, dataType, query, update)
}