Move out consul sync/lock and store. Move data/store to store

store/etcd/etcd.go

@@ -0,0 +1,91 @@
+// Package etcd is an etcd v3 implementation of kv
+package etcd
+
+import (
+	"context"
+	"log"
+
+	client "github.com/coreos/etcd/clientv3"
+	"github.com/micro/go-micro/config/options"
+	"github.com/micro/go-micro/store"
+)
+
+type ekv struct {
+	options.Options
+	kv client.KV
+}
+
+func (e *ekv) Read(key string) (*store.Record, error) {
+	keyval, err := e.kv.Get(context.Background(), key)
+	if err != nil {
+		return nil, err
+	}
+
+	if keyval == nil || len(keyval.Kvs) == 0 {
+		return nil, store.ErrNotFound
+	}
+
+	return &store.Record{
+		Key:   string(keyval.Kvs[0].Key),
+		Value: keyval.Kvs[0].Value,
+	}, nil
+}
+
+func (e *ekv) Delete(key string) error {
+	_, err := e.kv.Delete(context.Background(), key)
+	return err
+}
+
+func (e *ekv) Write(record *store.Record) error {
+	_, err := e.kv.Put(context.Background(), record.Key, string(record.Value))
+	return err
+}
+
+func (e *ekv) Dump() ([]*store.Record, error) {
+	keyval, err := e.kv.Get(context.Background(), "/", client.WithPrefix())
+	if err != nil {
+		return nil, err
+	}
+	var vals []*store.Record
+	if keyval == nil || len(keyval.Kvs) == 0 {
+		return vals, nil
+	}
+	for _, keyv := range keyval.Kvs {
+		vals = append(vals, &store.Record{
+			Key:   string(keyv.Key),
+			Value: keyv.Value,
+		})
+	}
+	return vals, nil
+}
+
+func (e *ekv) String() string {
+	return "etcd"
+}
+
+func NewStore(opts ...options.Option) store.Store {
+	options := options.NewOptions(opts...)
+
+	var endpoints []string
+
+	if e, ok := options.Values().Get("store.nodes"); ok {
+		endpoints = e.([]string)
+	}
+
+	if len(endpoints) == 0 {
+		endpoints = []string{"http://127.0.0.1:2379"}
+	}
+
+	// TODO: parse addresses
+	c, err := client.New(client.Config{
+		Endpoints: endpoints,
+	})
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	return &ekv{
+		Options: options,
+		kv:      client.NewKV(c),
+	}
+}

store/memory/memory.go

@@ -0,0 +1,108 @@
+// Package memory is a in-memory store store
+package memory
+
+import (
+	"sync"
+	"time"
+
+	"github.com/micro/go-micro/config/options"
+	"github.com/micro/go-micro/store"
+)
+
+type memoryStore struct {
+	options.Options
+
+	sync.RWMutex
+	values map[string]*memoryRecord
+}
+
+type memoryRecord struct {
+	r *store.Record
+	c time.Time
+}
+
+func (m *memoryStore) Dump() ([]*store.Record, error) {
+	m.RLock()
+	defer m.RUnlock()
+
+	var values []*store.Record
+
+	for _, v := range m.values {
+		// get expiry
+		d := v.r.Expiry
+		t := time.Since(v.c)
+
+		if d > time.Duration(0) {
+			// expired
+			if t > d {
+				continue
+			}
+			// update expiry
+			v.r.Expiry -= t
+			v.c = time.Now()
+		}
+
+		values = append(values, v.r)
+	}
+
+	return values, nil
+}
+
+func (m *memoryStore) Read(key string) (*store.Record, error) {
+	m.RLock()
+	defer m.RUnlock()
+
+	v, ok := m.values[key]
+	if !ok {
+		return nil, store.ErrNotFound
+	}
+
+	// get expiry
+	d := v.r.Expiry
+	t := time.Since(v.c)
+
+	// expired
+	if d > time.Duration(0) {
+		if t > d {
+			return nil, store.ErrNotFound
+		}
+		// update expiry
+		v.r.Expiry -= t
+		v.c = time.Now()
+	}
+
+	return v.r, nil
+}
+
+func (m *memoryStore) Write(r *store.Record) error {
+	m.Lock()
+	defer m.Unlock()
+
+	// set the record
+	m.values[r.Key] = &memoryRecord{
+		r: r,
+		c: time.Now(),
+	}
+
+	return nil
+}
+
+func (m *memoryStore) Delete(key string) error {
+	m.Lock()
+	defer m.Unlock()
+
+	// delete the value
+	delete(m.values, key)
+
+	return nil
+}
+
+// NewStore returns a new store.Store
+func NewStore(opts ...options.Option) store.Store {
+	options := options.NewOptions(opts...)
+
+	return &memoryStore{
+		Options: options,
+		values:  make(map[string]*memoryRecord),
+	}
+}

store/memory/memory_test.go

@@ -0,0 +1,37 @@
+package memory
+
+import (
+	"testing"
+	"time"
+
+	"github.com/micro/go-micro/store"
+)
+
+func TestReadRecordExpire(t *testing.T) {
+	s := NewStore()
+
+	var (
+		key    = "foo"
+		expire = 100 * time.Millisecond
+	)
+	rec := &store.Record{
+		Key:    key,
+		Value:  nil,
+		Expiry: expire,
+	}
+	s.Write(rec)
+
+	rrec, err := s.Read(key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if rrec.Expiry >= expire {
+		t.Fatal("expiry of read record is not changed")
+	}
+
+	time.Sleep(expire)
+
+	if _, err := s.Read(key); err != store.ErrNotFound {
+		t.Fatal("expire elapsed, but key still accessable")
+	}
+}

store/options.go

@@ -0,0 +1,15 @@
+package store
+
+import (
+	"github.com/micro/go-micro/config/options"
+)
+
+// Set the nodes used to back the store
+func Nodes(a ...string) options.Option {
+	return options.WithValue("store.nodes", a)
+}
+
+// Prefix sets a prefix to any key ids used
+func Prefix(p string) options.Option {
+	return options.WithValue("store.prefix", p)
+}

store/store.go

@@ -0,0 +1,34 @@
+// Package store is an interface for distribute data storage.
+package store
+
+import (
+	"errors"
+	"time"
+
+	"github.com/micro/go-micro/config/options"
+)
+
+var (
+	ErrNotFound = errors.New("not found")
+)
+
+// Store is a data storage interface
+type Store 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
+}