Move data to store
This commit is contained in:
Asim Aslam
@@ -1,96 +0,0 @@
|
||||
// Package consul is a consul implementation of kv
|
||||
package consul
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net"
|
||||
|
||||
"github.com/hashicorp/consul/api"
|
||||
"github.com/micro/go-micro/data"
|
||||
"github.com/micro/go-micro/options"
|
||||
)
|
||||
|
||||
type ckv struct {
|
||||
options.Options
|
||||
client *api.Client
|
||||
}
|
||||
|
||||
func (c *ckv) Read(key string) (*data.Record, error) {
|
||||
keyval, _, err := c.client.KV().Get(key, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if keyval == nil {
|
||||
return nil, data.ErrNotFound
|
||||
}
|
||||
|
||||
return &data.Record{
|
||||
Key: keyval.Key,
|
||||
Value: keyval.Value,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *ckv) Delete(key string) error {
|
||||
_, err := c.client.KV().Delete(key, nil)
|
||||
return err
|
||||
}
|
||||
|
||||
func (c *ckv) Write(record *data.Record) error {
|
||||
_, err := c.client.KV().Put(&api.KVPair{
|
||||
Key: record.Key,
|
||||
Value: record.Value,
|
||||
}, nil)
|
||||
return err
|
||||
}
|
||||
|
||||
func (c *ckv) Dump() ([]*data.Record, error) {
|
||||
keyval, _, err := c.client.KV().List("/", nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if keyval == nil {
|
||||
return nil, data.ErrNotFound
|
||||
}
|
||||
var vals []*data.Record
|
||||
for _, keyv := range keyval {
|
||||
vals = append(vals, &data.Record{
|
||||
Key: keyv.Key,
|
||||
Value: keyv.Value,
|
||||
})
|
||||
}
|
||||
return vals, nil
|
||||
}
|
||||
|
||||
func (c *ckv) String() string {
|
||||
return "consul"
|
||||
}
|
||||
|
||||
func NewData(opts ...options.Option) data.Data {
|
||||
options := options.NewOptions(opts...)
|
||||
config := api.DefaultConfig()
|
||||
|
||||
var nodes []string
|
||||
|
||||
if n, ok := options.Values().Get("data.nodes"); ok {
|
||||
nodes = n.([]string)
|
||||
}
|
||||
|
||||
// set host
|
||||
if len(nodes) > 0 {
|
||||
addr, port, err := net.SplitHostPort(nodes[0])
|
||||
if ae, ok := err.(*net.AddrError); ok && ae.Err == "missing port in address" {
|
||||
port = "8500"
|
||||
config.Address = fmt.Sprintf("%s:%s", nodes[0], port)
|
||||
} else if err == nil {
|
||||
config.Address = fmt.Sprintf("%s:%s", addr, port)
|
||||
}
|
||||
}
|
||||
|
||||
client, _ := api.NewClient(config)
|
||||
|
||||
return &ckv{
|
||||
Options: options,
|
||||
client: client,
|
||||
}
|
||||
}
|
||||
34
data/data.go
34
data/data.go
@@ -1,34 +0,0 @@
|
||||
// Package data is an interface for distribute data storage.
|
||||
package data
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"github.com/micro/go-micro/options"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrNotFound = errors.New("not found")
|
||||
)
|
||||
|
||||
// Data is a data storage interface
|
||||
type Data interface {
|
||||
// embed options
|
||||
options.Options
|
||||
// Dump the known records
|
||||
Dump() ([]*Record, error)
|
||||
// Read a record with key
|
||||
Read(key string) (*Record, error)
|
||||
// Write a record
|
||||
Write(r *Record) error
|
||||
// Delete a record with key
|
||||
Delete(key string) error
|
||||
}
|
||||
|
||||
// Record represents a data record
|
||||
type Record struct {
|
||||
Key string
|
||||
Value []byte
|
||||
Expiry time.Duration
|
||||
}
|
||||
@@ -1,97 +0,0 @@
|
||||
// Package memory is a in-memory data store
|
||||
package memory
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/micro/go-micro/data"
|
||||
"github.com/micro/go-micro/options"
|
||||
)
|
||||
|
||||
type memoryData struct {
|
||||
options.Options
|
||||
|
||||
sync.RWMutex
|
||||
values map[string]*memoryRecord
|
||||
}
|
||||
|
||||
type memoryRecord struct {
|
||||
r *data.Record
|
||||
c time.Time
|
||||
}
|
||||
|
||||
func (m *memoryData) Dump() ([]*data.Record, error) {
|
||||
m.RLock()
|
||||
defer m.RUnlock()
|
||||
|
||||
var values []*data.Record
|
||||
|
||||
for _, v := range m.values {
|
||||
// get expiry
|
||||
d := v.r.Expiry
|
||||
t := time.Since(v.c)
|
||||
|
||||
// expired
|
||||
if d > time.Duration(0) && t > d {
|
||||
continue
|
||||
}
|
||||
values = append(values, v.r)
|
||||
}
|
||||
|
||||
return values, nil
|
||||
}
|
||||
|
||||
func (m *memoryData) Read(key string) (*data.Record, error) {
|
||||
m.RLock()
|
||||
defer m.RUnlock()
|
||||
|
||||
v, ok := m.values[key]
|
||||
if !ok {
|
||||
return nil, data.ErrNotFound
|
||||
}
|
||||
|
||||
// get expiry
|
||||
d := v.r.Expiry
|
||||
t := time.Since(v.c)
|
||||
|
||||
// expired
|
||||
if d > time.Duration(0) && t > d {
|
||||
return nil, data.ErrNotFound
|
||||
}
|
||||
|
||||
return v.r, nil
|
||||
}
|
||||
|
||||
func (m *memoryData) Write(r *data.Record) error {
|
||||
m.Lock()
|
||||
defer m.Unlock()
|
||||
|
||||
// set the record
|
||||
m.values[r.Key] = &memoryRecord{
|
||||
r: r,
|
||||
c: time.Now(),
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *memoryData) Delete(key string) error {
|
||||
m.Lock()
|
||||
defer m.Unlock()
|
||||
|
||||
// delete the value
|
||||
delete(m.values, key)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// NewData returns a new data.Data
|
||||
func NewData(opts ...options.Option) data.Data {
|
||||
options := options.NewOptions(opts...)
|
||||
|
||||
return &memoryData{
|
||||
Options: options,
|
||||
values: make(map[string]*memoryRecord),
|
||||
}
|
||||
}
|
||||
@@ -1,15 +0,0 @@
|
||||
package data
|
||||
|
||||
import (
|
||||
"github.com/micro/go-micro/options"
|
||||
)
|
||||
|
||||
// Set the nodes used to back the data
|
||||
func Nodes(a ...string) options.Option {
|
||||
return options.WithValue("data.nodes", a)
|
||||
}
|
||||
|
||||
// Prefix sets a prefix to any key ids used
|
||||
func Prefix(p string) options.Option {
|
||||
return options.WithValue("data.prefix", p)
|
||||
}
|
||||
Reference in New Issue
Block a user