Add opts to service proto (#1517)

* Add opts to service proto

* Support database/table opts
This commit is contained in:
Asim Aslam 2020-04-30 22:51:25 +01:00 committed by Vasiliy Tolstov
parent 82496bae86
commit 71dc4459d2

190
file.go
View File

@ -7,10 +7,10 @@ import (
"path/filepath" "path/filepath"
"sort" "sort"
"strings" "strings"
"sync"
"time" "time"
"github.com/micro/go-micro/v2/store" "github.com/micro/go-micro/v2/store"
"github.com/pkg/errors"
bolt "go.etcd.io/bbolt" bolt "go.etcd.io/bbolt"
) )
@ -29,18 +29,25 @@ var (
// NewStore returns a memory store // NewStore returns a memory store
func NewStore(opts ...store.Option) store.Store { func NewStore(opts ...store.Option) store.Store {
s := &fileStore{} s := &fileStore{
handles: make(map[string]*fileHandle),
}
s.init(opts...) s.init(opts...)
return s return s
} }
type fileStore struct { type fileStore struct {
options store.Options options store.Options
dir string dir string
fileName string
dbPath string
// the database handle // the database handle
db *bolt.DB sync.RWMutex
handles map[string]*fileHandle
}
type fileHandle struct {
key string
db *bolt.DB
} }
// record stored by us // record stored by us
@ -50,8 +57,12 @@ type record struct {
ExpiresAt time.Time ExpiresAt time.Time
} }
func (m *fileStore) delete(key string) error { func key(database, table string) string {
return m.db.Update(func(tx *bolt.Tx) error { return database + ":" + table
}
func (m *fileStore) delete(fd *fileHandle, key string) error {
return fd.db.Update(func(tx *bolt.Tx) error {
b := tx.Bucket([]byte(dataBucket)) b := tx.Bucket([]byte(dataBucket))
if b == nil { if b == nil {
return nil return nil
@ -76,42 +87,63 @@ func (m *fileStore) init(opts ...store.Option) error {
// create a directory /tmp/micro // create a directory /tmp/micro
dir := filepath.Join(DefaultDir, m.options.Database) dir := filepath.Join(DefaultDir, m.options.Database)
// create the database handle
fname := m.options.Table + ".db"
// Ignoring this as the folder might exist. // Ignoring this as the folder might exist.
// Reads/Writes updates will return with sensible error messages // Reads/Writes updates will return with sensible error messages
// about the dir not existing in case this cannot create the path anyway // about the dir not existing in case this cannot create the path anyway
os.MkdirAll(dir, 0700) os.MkdirAll(dir, 0700)
m.dir = dir return nil
m.fileName = fname
m.dbPath = filepath.Join(dir, fname)
// close existing handle
if m.db != nil {
m.db.Close()
}
// create new db handle
db, err := bolt.Open(m.dbPath, 0700, &bolt.Options{Timeout: 5 * time.Second})
if err != nil {
return err
}
// set the new db
m.db = db
// create the table
return db.Update(func(tx *bolt.Tx) error {
_, err := tx.CreateBucketIfNotExists([]byte(dataBucket))
return err
})
} }
func (m *fileStore) list(limit, offset uint) []string { func (f *fileStore) getDB(database, table string) (*fileHandle, error) {
if len(database) == 0 {
database = f.options.Database
}
if len(table) == 0 {
table = f.options.Table
}
k := key(database, table)
f.RLock()
fd, ok := f.handles[k]
f.RUnlock()
// return the file handle
if ok {
return fd, nil
}
// create a directory /tmp/micro
dir := filepath.Join(DefaultDir, database)
// create the database handle
fname := table + ".db"
// make the dir
os.MkdirAll(dir, 0700)
// database path
dbPath := filepath.Join(dir, fname)
// create new db handle
db, err := bolt.Open(dbPath, 0700, &bolt.Options{Timeout: 5 * time.Second})
if err != nil {
return nil, err
}
f.Lock()
fd = &fileHandle{
key: k,
db: db,
}
f.handles[k] = fd
f.Unlock()
return fd, nil
}
func (m *fileStore) list(fd *fileHandle, limit, offset uint) []string {
var allItems []string var allItems []string
m.db.View(func(tx *bolt.Tx) error { fd.db.View(func(tx *bolt.Tx) error {
b := tx.Bucket([]byte(dataBucket)) b := tx.Bucket([]byte(dataBucket))
// nothing to read // nothing to read
if b == nil { if b == nil {
@ -162,10 +194,10 @@ func (m *fileStore) list(limit, offset uint) []string {
return allKeys return allKeys
} }
func (m *fileStore) get(k string) (*store.Record, error) { func (m *fileStore) get(fd *fileHandle, k string) (*store.Record, error) {
var value []byte var value []byte
m.db.View(func(tx *bolt.Tx) error { fd.db.View(func(tx *bolt.Tx) error {
// @todo this is still very experimental... // @todo this is still very experimental...
b := tx.Bucket([]byte(dataBucket)) b := tx.Bucket([]byte(dataBucket))
if b == nil { if b == nil {
@ -200,7 +232,7 @@ func (m *fileStore) get(k string) (*store.Record, error) {
return newRecord, nil return newRecord, nil
} }
func (m *fileStore) set(r *store.Record) error { func (m *fileStore) set(fd *fileHandle, r *store.Record) error {
// copy the incoming record and then // copy the incoming record and then
// convert the expiry in to a hard timestamp // convert the expiry in to a hard timestamp
item := &record{} item := &record{}
@ -213,7 +245,7 @@ func (m *fileStore) set(r *store.Record) error {
// marshal the data // marshal the data
data, _ := json.Marshal(item) data, _ := json.Marshal(item)
return m.db.Update(func(tx *bolt.Tx) error { return fd.db.Update(func(tx *bolt.Tx) error {
b := tx.Bucket([]byte(dataBucket)) b := tx.Bucket([]byte(dataBucket))
if b == nil { if b == nil {
var err error var err error
@ -226,53 +258,63 @@ func (m *fileStore) set(r *store.Record) error {
}) })
} }
func (m *fileStore) Close() error { func (f *fileStore) Close() error {
if m.db != nil { f.Lock()
return m.db.Close() defer f.Unlock()
for k, v := range f.handles {
v.db.Close()
delete(f.handles, k)
} }
return nil return nil
} }
func (m *fileStore) Init(opts ...store.Option) error { func (f *fileStore) Init(opts ...store.Option) error {
return m.init(opts...) return f.init(opts...)
} }
func (m *fileStore) Delete(key string, opts ...store.DeleteOption) error { func (m *fileStore) Delete(key string, opts ...store.DeleteOption) error {
deleteOptions := store.DeleteOptions{} var deleteOptions store.DeleteOptions
for _, o := range opts { for _, o := range opts {
o(&deleteOptions) o(&deleteOptions)
} }
return m.delete(key)
fd, err := m.getDB(deleteOptions.Database, deleteOptions.Table)
if err != nil {
return err
}
return m.delete(fd, key)
} }
func (m *fileStore) Read(key string, opts ...store.ReadOption) ([]*store.Record, error) { func (m *fileStore) Read(key string, opts ...store.ReadOption) ([]*store.Record, error) {
readOpts := store.ReadOptions{} var readOpts store.ReadOptions
for _, o := range opts { for _, o := range opts {
o(&readOpts) o(&readOpts)
} }
fd, err := m.getDB(readOpts.Database, readOpts.Table)
if err != nil {
return nil, err
}
var keys []string var keys []string
// Handle Prefix / suffix // Handle Prefix / suffix
// TODO: do range scan here rather than listing all keys // TODO: do range scan here rather than listing all keys
if readOpts.Prefix || readOpts.Suffix { if readOpts.Prefix || readOpts.Suffix {
var opts []store.ListOption // list the keys
if readOpts.Prefix { k := m.list(fd, readOpts.Limit, readOpts.Offset)
opts = append(opts, store.ListPrefix(key))
}
if readOpts.Suffix {
opts = append(opts, store.ListSuffix(key))
}
opts = append(opts, store.ListLimit(readOpts.Limit)) // check for prefix and suffix
opts = append(opts, store.ListOffset(readOpts.Offset)) for _, v := range k {
if readOpts.Prefix && !strings.HasPrefix(v, key) {
k, err := m.List(opts...) continue
if err != nil { }
return nil, errors.Wrap(err, "FileStore: Read couldn't List()") if readOpts.Suffix && !strings.HasSuffix(v, key) {
continue
}
keys = append(keys, v)
} }
keys = k
} else { } else {
keys = []string{key} keys = []string{key}
} }
@ -280,7 +322,7 @@ func (m *fileStore) Read(key string, opts ...store.ReadOption) ([]*store.Record,
var results []*store.Record var results []*store.Record
for _, k := range keys { for _, k := range keys {
r, err := m.get(k) r, err := m.get(fd, k)
if err != nil { if err != nil {
return results, err return results, err
} }
@ -291,11 +333,16 @@ func (m *fileStore) Read(key string, opts ...store.ReadOption) ([]*store.Record,
} }
func (m *fileStore) Write(r *store.Record, opts ...store.WriteOption) error { func (m *fileStore) Write(r *store.Record, opts ...store.WriteOption) error {
writeOpts := store.WriteOptions{} var writeOpts store.WriteOptions
for _, o := range opts { for _, o := range opts {
o(&writeOpts) o(&writeOpts)
} }
fd, err := m.getDB(writeOpts.Database, writeOpts.Table)
if err != nil {
return err
}
if len(opts) > 0 { if len(opts) > 0 {
// Copy the record before applying options, or the incoming record will be mutated // Copy the record before applying options, or the incoming record will be mutated
newRecord := store.Record{} newRecord := store.Record{}
@ -310,10 +357,10 @@ func (m *fileStore) Write(r *store.Record, opts ...store.WriteOption) error {
newRecord.Expiry = writeOpts.TTL newRecord.Expiry = writeOpts.TTL
} }
return m.set(&newRecord) return m.set(fd, &newRecord)
} }
return m.set(r) return m.set(fd, r)
} }
func (m *fileStore) Options() store.Options { func (m *fileStore) Options() store.Options {
@ -321,14 +368,19 @@ func (m *fileStore) Options() store.Options {
} }
func (m *fileStore) List(opts ...store.ListOption) ([]string, error) { func (m *fileStore) List(opts ...store.ListOption) ([]string, error) {
listOptions := store.ListOptions{} var listOptions store.ListOptions
for _, o := range opts { for _, o := range opts {
o(&listOptions) o(&listOptions)
} }
fd, err := m.getDB(listOptions.Database, listOptions.Table)
if err != nil {
return nil, err
}
// TODO apply prefix/suffix in range query // TODO apply prefix/suffix in range query
allKeys := m.list(listOptions.Limit, listOptions.Offset) allKeys := m.list(fd, listOptions.Limit, listOptions.Offset)
if len(listOptions.Prefix) > 0 { if len(listOptions.Prefix) > 0 {
var prefixKeys []string var prefixKeys []string