Finish implementation of the KV

3 files changed, 697 insertions, 0 deletions

diff --git a/plugins/kv/memcached/config.go b/plugins/kv/memcached/config.go
new file mode 100644
index 00000000..62f29ef2
--- /dev/null
+++ b/plugins/kv/memcached/config.go
@@ -0,0 +1,10 @@
+package memcached
+
+type Config struct {
+	// Addr is url for memcached, 11211 port is used by default
+	Addr []string
+}
+
+func (s *Config) InitDefaults() {
+	s.Addr = []string{"localhost:11211"} // default url for memcached     // init logger
+}
diff --git a/plugins/kv/memcached/plugin.go b/plugins/kv/memcached/plugin.go
new file mode 100644
index 00000000..bd0a207d
--- /dev/null
+++ b/plugins/kv/memcached/plugin.go
@@ -0,0 +1,243 @@
+package memcached
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	"github.com/bradfitz/gomemcache/memcache"
+	"github.com/spiral/errors"
+	"github.com/spiral/roadrunner/v2/plugins/config"
+	"github.com/spiral/roadrunner/v2/plugins/kv"
+	"github.com/spiral/roadrunner/v2/plugins/logger"
+)
+
+const PluginName = "memcached"
+
+var EmptyItem = kv.Item{}
+
+type Plugin struct {
+	// config
+	cfg *Config
+	// logger
+	log logger.Logger
+	// memcached client
+	client *memcache.Client
+}
+
+// NewMemcachedClient returns a memcache client using the provided server(s)
+// with equal weight. If a server is listed multiple times,
+// it gets a proportional amount of weight.
+func NewMemcachedClient(url string) kv.Storage {
+	m := memcache.New(url)
+	return &Plugin{
+		client: m,
+	}
+}
+
+func (s *Plugin) Init(log logger.Logger, cfg config.Configurer) error {
+	const op = errors.Op("memcached init")
+	s.cfg = &Config{}
+	s.cfg.InitDefaults()
+	err := cfg.UnmarshalKey(PluginName, &s.cfg)
+	if err != nil {
+		return errors.E(op, err)
+	}
+	s.log = log
+	return nil
+}
+
+func (s *Plugin) Serve() chan error {
+	errCh := make(chan error, 1)
+	s.client = memcache.New(s.cfg.Addr...)
+	return errCh
+}
+
+// Memcached has no stop/close or smt similar to close the connection
+func (s *Plugin) Stop() error {
+	return nil
+}
+
+// Has checks the key for existence
+func (s Plugin) Has(ctx context.Context, keys ...string) (map[string]bool, error) {
+	const op = errors.Op("memcached Has")
+	if keys == nil {
+		return nil, errors.E(op, errors.NoKeys)
+	}
+	m := make(map[string]bool, len(keys))
+	for _, key := range keys {
+		keyTrimmed := strings.TrimSpace(key)
+		if keyTrimmed == "" {
+			return nil, errors.E(op, errors.EmptyKey)
+		}
+		exist, err := s.client.Get(key)
+		// ErrCacheMiss means that a Get failed because the item wasn't present.
+		if err != nil && err != memcache.ErrCacheMiss {
+			return nil, err
+		}
+		if exist != nil {
+			m[key] = true
+		}
+	}
+	return m, nil
+}
+
+// Get gets the item for the given key. ErrCacheMiss is returned for a
+// memcache cache miss. The key must be at most 250 bytes in length.
+func (s Plugin) Get(ctx context.Context, key string) ([]byte, error) {
+	const op = errors.Op("memcached Get")
+	// to get cases like "  "
+	keyTrimmed := strings.TrimSpace(key)
+	if keyTrimmed == "" {
+		return nil, errors.E(op, errors.EmptyKey)
+	}
+	data, err := s.client.Get(key)
+	// ErrCacheMiss means that a Get failed because the item wasn't present.
+	if err != nil && err != memcache.ErrCacheMiss {
+		return nil, err
+	}
+	if data != nil {
+		// return the value by the key
+		return data.Value, nil
+	}
+	// data is nil by some reason and error also nil
+	return nil, nil
+}
+
+// return map with key -- string
+// and map value as value -- []byte
+func (s Plugin) MGet(ctx context.Context, keys ...string) (map[string]interface{}, error) {
+	const op = errors.Op("memcached MGet")
+	if keys == nil {
+		return nil, errors.E(op, errors.NoKeys)
+	}
+
+	// should not be empty keys
+	for _, key := range keys {
+		keyTrimmed := strings.TrimSpace(key)
+		if keyTrimmed == "" {
+			return nil, errors.E(op, errors.EmptyKey)
+		}
+	}
+
+	m := make(map[string]interface{}, len(keys))
+	for _, key := range keys {
+		// Here also MultiGet
+		data, err := s.client.Get(key)
+		// ErrCacheMiss means that a Get failed because the item wasn't present.
+		if err != nil && err != memcache.ErrCacheMiss {
+			return nil, err
+		}
+		if data != nil {
+			m[key] = data.Value
+		}
+	}
+
+	return m, nil
+}
+
+// Set sets the KV pairs. Keys should be 250 bytes maximum
+// TTL:
+// Expiration is the cache expiration time, in seconds: either a relative
+// time from now (up to 1 month), or an absolute Unix epoch time.
+// Zero means the Item has no expiration time.
+func (s Plugin) Set(ctx context.Context, items ...kv.Item) error {
+	const op = errors.Op("memcached Set")
+	if items == nil {
+		return errors.E(op, errors.NoKeys)
+	}
+
+	for i := range items {
+		if items[i] == EmptyItem {
+			return errors.E(op, errors.EmptyItem)
+		}
+
+		// pre-allocate item
+		memcachedItem := &memcache.Item{
+			Key: items[i].Key,
+			// unsafe convert
+			Value: []byte(items[i].Value),
+			Flags: 0,
+		}
+
+		// add additional TTL in case of TTL isn't empty
+		if items[i].TTL != "" {
+			// verify the TTL
+			t, err := time.Parse(time.RFC3339, items[i].TTL)
+			if err != nil {
+				return err
+			}
+			memcachedItem.Expiration = int32(t.Unix())
+		}
+
+		err := s.client.Set(memcachedItem)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// Expiration is the cache expiration time, in seconds: either a relative
+// time from now (up to 1 month), or an absolute Unix epoch time.
+// Zero means the Item has no expiration time.
+func (s Plugin) MExpire(ctx context.Context, items ...kv.Item) error {
+	const op = errors.Op("memcached MExpire")
+	for _, item := range items {
+		if item.TTL == "" || strings.TrimSpace(item.Key) == "" {
+			return errors.E(op, errors.Str("should set timeout and at least one key"))
+		}
+
+		// verify provided TTL
+		t, err := time.Parse(time.RFC3339, item.TTL)
+		if err != nil {
+			return err
+		}
+
+		// Touch updates the expiry for the given key. The seconds parameter is either
+		// a Unix timestamp or, if seconds is less than 1 month, the number of seconds
+		// into the future at which time the item will expire. Zero means the item has
+		// no expiration time. ErrCacheMiss is returned if the key is not in the cache.
+		// The key must be at most 250 bytes in length.
+		err = s.client.Touch(item.Key, int32(t.Unix()))
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// return time in seconds (int32) for a given keys
+func (s Plugin) TTL(ctx context.Context, keys ...string) (map[string]interface{}, error) {
+	const op = errors.Op("memcached HTTLas")
+	return nil, errors.E(op, errors.Str("not valid request for memcached, see https://github.com/memcached/memcached/issues/239"))
+}
+
+func (s Plugin) Delete(ctx context.Context, keys ...string) error {
+	const op = errors.Op("memcached Has")
+	if keys == nil {
+		return errors.E(op, errors.NoKeys)
+	}
+
+	// should not be empty keys
+	for _, key := range keys {
+		keyTrimmed := strings.TrimSpace(key)
+		if keyTrimmed == "" {
+			return errors.E(op, errors.EmptyKey)
+		}
+	}
+
+	for _, key := range keys {
+		err := s.client.Delete(key)
+		// ErrCacheMiss means that a Get failed because the item wasn't present.
+		if err != nil && err != memcache.ErrCacheMiss {
+			return err
+		}
+	}
+	return nil
+}
+
+func (s Plugin) Close() error {
+	return nil
+}
diff --git a/plugins/kv/memcached/storage_test.go b/plugins/kv/memcached/storage_test.go
new file mode 100644
index 00000000..4b59bbd0
--- /dev/null
+++ b/plugins/kv/memcached/storage_test.go
@@ -0,0 +1,444 @@
+package memcached
+
+import (
+	"context"
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/spiral/roadrunner/v2/plugins/kv"
+	"github.com/stretchr/testify/assert"
+)
+
+func initStorage() kv.Storage {
+	return NewMemcachedClient("localhost:11211")
+}
+
+func cleanup(t *testing.T, s kv.Storage, keys ...string) {
+	err := s.Delete(context.Background(), keys...)
+	if err != nil {
+		t.Fatalf("error during cleanup: %s", err.Error())
+	}
+}
+
+func TestStorage_Has(t *testing.T) {
+	s := initStorage()
+
+	ctx := context.Background()
+
+	v, err := s.Has(ctx, "key")
+	assert.NoError(t, err)
+	assert.False(t, v["key"])
+}
+
+func TestStorage_Has_Set_Has(t *testing.T) {
+	s := initStorage()
+	defer func() {
+		cleanup(t, s, "key", "key2")
+		if err := s.Close(); err != nil {
+			panic(err)
+		}
+	}()
+
+	ctx := context.Background()
+	v, err := s.Has(ctx, "key")
+	assert.NoError(t, err)
+	// no such key
+	assert.False(t, v["key"])
+
+	assert.NoError(t, s.Set(ctx, kv.Item{
+		Key:   "key",
+		Value: "hello world",
+		TTL:   "",
+	}, kv.Item{
+		Key:   "key2",
+		Value: "hello world",
+		TTL:   "",
+	}))
+
+	v, err = s.Has(ctx, "key", "key2")
+	assert.NoError(t, err)
+	// no such key
+	assert.True(t, v["key"])
+	assert.True(t, v["key2"])
+}
+
+func TestStorage_Has_Set_MGet(t *testing.T) {
+	s := initStorage()
+	defer func() {
+		cleanup(t, s, "key", "key2")
+		if err := s.Close(); err != nil {
+			panic(err)
+		}
+	}()
+
+	ctx := context.Background()
+	v, err := s.Has(ctx, "key")
+	assert.NoError(t, err)
+	// no such key
+	assert.False(t, v["key"])
+
+	assert.NoError(t, s.Set(ctx, kv.Item{
+		Key:   "key",
+		Value: "hello world",
+		TTL:   "",
+	}, kv.Item{
+		Key:   "key2",
+		Value: "hello world",
+		TTL:   "",
+	}))
+
+	v, err = s.Has(ctx, "key", "key2")
+	assert.NoError(t, err)
+	// no such key
+	assert.True(t, v["key"])
+	assert.True(t, v["key2"])
+
+	res, err := s.MGet(ctx, "key", "key2")
+	assert.NoError(t, err)
+	assert.Len(t, res, 2)
+}
+
+func TestStorage_Has_Set_Get(t *testing.T) {
+	s := initStorage()
+	defer func() {
+		cleanup(t, s, "key", "key2")
+		if err := s.Close(); err != nil {
+			panic(err)
+		}
+	}()
+
+	ctx := context.Background()
+	v, err := s.Has(ctx, "key")
+	assert.NoError(t, err)
+	// no such key
+	assert.False(t, v["key"])
+
+	assert.NoError(t, s.Set(ctx, kv.Item{
+		Key:   "key",
+		Value: "hello world",
+		TTL:   "",
+	}, kv.Item{
+		Key:   "key2",
+		Value: "hello world",
+		TTL:   "",
+	}))
+
+	v, err = s.Has(ctx, "key", "key2")
+	assert.NoError(t, err)
+	// no such key
+	assert.True(t, v["key"])
+	assert.True(t, v["key2"])
+
+	res, err := s.Get(ctx, "key")
+	assert.NoError(t, err)
+
+	if string(res) != "hello world" {
+		t.Fatal("wrong value by key")
+	}
+}
+
+func TestStorage_Set_Del_Get(t *testing.T) {
+	s := initStorage()
+	defer func() {
+		cleanup(t, s, "key", "key2")
+		if err := s.Close(); err != nil {
+			panic(err)
+		}
+	}()
+
+	ctx := context.Background()
+	v, err := s.Has(ctx, "key")
+	assert.NoError(t, err)
+	// no such key
+	assert.False(t, v["key"])
+
+	assert.NoError(t, s.Set(ctx, kv.Item{
+		Key:   "key",
+		Value: "hello world",
+		TTL:   "",
+	}, kv.Item{
+		Key:   "key2",
+		Value: "hello world",
+		TTL:   "",
+	}))
+
+	v, err = s.Has(ctx, "key", "key2")
+	assert.NoError(t, err)
+	// no such key
+	assert.True(t, v["key"])
+	assert.True(t, v["key2"])
+
+	// check that keys are present
+	res, err := s.MGet(ctx, "key", "key2")
+	assert.NoError(t, err)
+	assert.Len(t, res, 2)
+
+	assert.NoError(t, s.Delete(ctx, "key", "key2"))
+	// check that keys are not present
+	res, err = s.MGet(ctx, "key", "key2")
+	assert.NoError(t, err)
+	assert.Len(t, res, 0)
+}
+
+func TestStorage_Set_GetM(t *testing.T) {
+	s := initStorage()
+	ctx := context.Background()
+
+	defer func() {
+		cleanup(t, s, "key", "key2")
+
+		if err := s.Close(); err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	v, err := s.Has(ctx, "key")
+	assert.NoError(t, err)
+	assert.False(t, v["key"])
+
+	assert.NoError(t, s.Set(ctx, kv.Item{
+		Key:   "key",
+		Value: "hello world",
+		TTL:   "",
+	}, kv.Item{
+		Key:   "key2",
+		Value: "hello world",
+		TTL:   "",
+	}))
+
+	res, err := s.MGet(ctx, "key", "key2")
+	assert.NoError(t, err)
+	assert.Len(t, res, 2)
+}
+
+func TestStorage_MExpire_TTL(t *testing.T) {
+	s := initStorage()
+	ctx := context.Background()
+	defer func() {
+		cleanup(t, s, "key", "key2")
+		if err := s.Close(); err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	// ensure that storage is clean
+	v, err := s.Has(ctx, "key", "key2")
+	assert.NoError(t, err)
+	assert.False(t, v["key"])
+	assert.False(t, v["key2"])
+
+	assert.NoError(t, s.Set(ctx, kv.Item{
+		Key:   "key",
+		Value: "hello world",
+		TTL:   "",
+	},
+		kv.Item{
+			Key:   "key2",
+			Value: "hello world",
+			TTL:   "",
+		}))
+	// set timeout to 5 sec
+	nowPlusFive := time.Now().Add(time.Second * 5).Format(time.RFC3339)
+
+	i1 := kv.Item{
+		Key:   "key",
+		Value: "",
+		TTL:   nowPlusFive,
+	}
+	i2 := kv.Item{
+		Key:   "key2",
+		Value: "",
+		TTL:   nowPlusFive,
+	}
+	assert.NoError(t, s.MExpire(ctx, i1, i2))
+
+	time.Sleep(time.Second * 6)
+
+	// ensure that storage is clean
+	v, err = s.Has(ctx, "key", "key2")
+	assert.NoError(t, err)
+	assert.False(t, v["key"])
+	assert.False(t, v["key2"])
+}
+
+func TestNilAndWrongArgs(t *testing.T) {
+	s := initStorage()
+	ctx := context.Background()
+	defer func() {
+		cleanup(t, s, "key")
+		if err := s.Close(); err != nil {
+			panic(err)
+		}
+	}()
+
+	// check
+	v, err := s.Has(ctx, "key")
+	assert.NoError(t, err)
+	assert.False(t, v["key"])
+
+	_, err = s.Has(ctx, "")
+	assert.Error(t, err)
+
+	_, err = s.Get(ctx, "")
+	assert.Error(t, err)
+
+	_, err = s.Get(ctx, " ")
+	assert.Error(t, err)
+
+	_, err = s.Get(ctx, "                 ")
+	assert.Error(t, err)
+
+	_, err = s.MGet(ctx, "key", "key2", "")
+	assert.Error(t, err)
+
+	_, err = s.MGet(ctx, "key", "key2", "   ")
+	assert.Error(t, err)
+
+	assert.Error(t, s.Set(ctx, kv.Item{}))
+
+	err = s.Delete(ctx, "")
+	assert.Error(t, err)
+
+	err = s.Delete(ctx, "key", "")
+	assert.Error(t, err)
+
+	err = s.Delete(ctx, "key", "     ")
+	assert.Error(t, err)
+
+	err = s.Delete(ctx, "key")
+	assert.NoError(t, err)
+}
+
+func TestStorage_SetExpire_TTL(t *testing.T) {
+	s := initStorage()
+	ctx := context.Background()
+	defer func() {
+		cleanup(t, s, "key", "key2")
+		if err := s.Close(); err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	// ensure that storage is clean
+	v, err := s.Has(ctx, "key", "key2")
+	assert.NoError(t, err)
+	assert.False(t, v["key"])
+	assert.False(t, v["key2"])
+
+	assert.NoError(t, s.Set(ctx, kv.Item{
+		Key:   "key",
+		Value: "hello world",
+		TTL:   "",
+	},
+		kv.Item{
+			Key:   "key2",
+			Value: "hello world",
+			TTL:   "",
+		}))
+
+	nowPlusFive := time.Now().Add(time.Second * 5).Format(time.RFC3339)
+
+	// set timeout to 5 sec
+	assert.NoError(t, s.Set(ctx, kv.Item{
+		Key:   "key",
+		Value: "value",
+		TTL:   nowPlusFive,
+	},
+		kv.Item{
+			Key:   "key2",
+			Value: "value",
+			TTL:   nowPlusFive,
+		}))
+
+	time.Sleep(time.Second * 6)
+
+	// ensure that storage is clean
+	v, err = s.Has(ctx, "key", "key2")
+	assert.NoError(t, err)
+	assert.False(t, v["key"])
+	assert.False(t, v["key2"])
+}
+
+func TestConcurrentReadWriteTransactions(t *testing.T) {
+	s := initStorage()
+	defer func() {
+		cleanup(t, s, "key", "key2")
+		if err := s.Close(); err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	ctx := context.Background()
+	v, err := s.Has(ctx, "key")
+	assert.NoError(t, err)
+	// no such key
+	assert.False(t, v["key"])
+
+	assert.NoError(t, s.Set(ctx, kv.Item{
+		Key:   "key",
+		Value: "hello world",
+		TTL:   "",
+	}, kv.Item{
+		Key:   "key2",
+		Value: "hello world",
+		TTL:   "",
+	}))
+
+	v, err = s.Has(ctx, "key", "key2")
+	assert.NoError(t, err)
+	// no such key
+	assert.True(t, v["key"])
+	assert.True(t, v["key2"])
+
+	wg := &sync.WaitGroup{}
+	wg.Add(3)
+
+	m := &sync.RWMutex{}
+	// concurrently set the keys
+	go func(s kv.Storage) {
+		defer wg.Done()
+		for i := 0; i <= 1000; i++ {
+			m.Lock()
+			// set is writable transaction
+			// it should stop readable
+			assert.NoError(t, s.Set(ctx, kv.Item{
+				Key:   "key" + strconv.Itoa(i),
+				Value: "hello world" + strconv.Itoa(i),
+				TTL:   "",
+			}, kv.Item{
+				Key:   "key2" + strconv.Itoa(i),
+				Value: "hello world" + strconv.Itoa(i),
+				TTL:   "",
+			}))
+			m.Unlock()
+		}
+	}(s)
+
+	// should be no errors
+	go func(s kv.Storage) {
+		defer wg.Done()
+		for i := 0; i <= 1000; i++ {
+			m.RLock()
+			v, err = s.Has(ctx, "key")
+			assert.NoError(t, err)
+			// no such key
+			assert.True(t, v["key"])
+			m.RUnlock()
+		}
+	}(s)
+
+	// should be no errors
+	go func(s kv.Storage) {
+		defer wg.Done()
+		for i := 0; i <= 1000; i++ {
+			m.Lock()
+			err = s.Delete(ctx, "key"+strconv.Itoa(i))
+			assert.NoError(t, err)
+			m.Unlock()
+		}
+	}(s)
+
+	wg.Wait()
+}