/
mem.go
71 lines (60 loc) · 1.36 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
package storage
import (
"context"
"reflect"
"sync"
"github.com/go-pg/pg/v10/orm"
"github.com/filecoin-project/lily/model"
)
func NewMemStorage(version model.Version) *MemStorage {
return &MemStorage{
Data: map[string][]interface{}{},
Version: version,
}
}
func NewMemStorageLatest() *MemStorage {
return NewMemStorage(LatestSchemaVersion())
}
type MemStorage struct {
// TODO parallel map?
Data map[string][]interface{}
DataMu sync.Mutex
Version model.Version
}
func (j *MemStorage) PersistModel(ctx context.Context, m interface{}) error {
if len(Models) == 0 {
return nil
}
value := reflect.ValueOf(m)
if value.Kind() == reflect.Ptr {
value = value.Elem()
}
switch value.Kind() {
case reflect.Slice, reflect.Array:
for i := 0; i < value.Len(); i++ {
if err := j.PersistModel(ctx, value.Index(i).Interface()); err != nil {
return err
}
}
return nil
case reflect.Struct:
q := orm.NewQuery(nil, m)
tm := q.TableModel()
n := tm.Table()
name := stripQuotes(n.SQLNameForSelects)
j.DataMu.Lock()
j.Data[name] = append(j.Data[name], m)
j.DataMu.Unlock()
return nil
default:
return ErrMarshalUnsupportedType
}
}
func (j *MemStorage) PersistBatch(ctx context.Context, ps ...model.Persistable) error {
for _, p := range ps {
if err := p.Persist(ctx, j, j.Version); err != nil {
return err
}
}
return nil
}