refines DAL from DAO

store/etcdv3/meta/etcd.go

@@ -74,11 +74,6 @@ func NewETCD(config types.EtcdConfig, embeddedStorage bool) (*ETCD, error) {
 	return &ETCD{cliv3: cliv3, config: config}, nil
 }
 
-// ClientV3 gets the raw ETCD client v3.
-func (e *ETCD) ClientV3() *clientv3.Client {
-	return e.cliv3.(*clientv3.Client)
-}
-
 // TerminateEmbededStorage terminate embedded storage
 func (e *ETCD) TerminateEmbededStorage() {
 	embedded.TerminateCluster()
@@ -87,7 +82,7 @@ func (e *ETCD) TerminateEmbededStorage() {
 // CreateLock create a lock instance
 func (e *ETCD) CreateLock(key string, ttl time.Duration) (lock.DistributedLock, error) {
 	lockKey := fmt.Sprintf("%s/%s", e.config.LockPrefix, key)
-	mutex, err := etcdlock.New(e.ClientV3(), lockKey, ttl)
+	mutex, err := etcdlock.New(e.cliv3.(*clientv3.Client), lockKey, ttl)
 	return mutex, err
 }
 
@@ -233,6 +228,21 @@ func (e *ETCD) BatchUpdate(ctx context.Context, data map[string]string, opts ...
 	return e.batchUpdate(ctx, data, opts...)
 }
 
+// Grant creates a new lease.
+func (e *ETCD) Grant(ctx context.Context, ttl int64) (*clientv3.LeaseGrantResponse, error) {
+	return e.cliv3.Grant(ctx, ttl)
+}
+
+// KeepAliveOnce keeps on a lease alive.
+func (e *ETCD) KeepAliveOnce(ctx context.Context, id clientv3.LeaseID) (*clientv3.LeaseKeepAliveResponse, error) {
+	return e.cliv3.KeepAliveOnce(ctx, id)
+}
+
+// Txn creates a new Txn
+func (e *ETCD) Txn(ctx context.Context) clientv3.Txn {
+	return e.cliv3.Txn(ctx)
+}
+
 func (e *ETCD) batchUpdate(ctx context.Context, data map[string]string, opts ...clientv3.OpOption) (*clientv3.TxnResponse, error) {
 	limit := map[string]map[string]string{}
 	for key := range data {

store/etcdv3/meta/etcd_test.go

@@ -52,6 +52,31 @@ func TestGetMultiFailedAsBatchGetError(t *testing.T) {
 	require.Nil(t, kvs)
 }
 
+func TestGrant(t *testing.T) {
+	e := NewMockedETCD(t)
+	expErr := fmt.Errorf("exp")
+	e.cliv3.(*mocks.ETCDClientV3).On("Grant", mock.Anything, mock.Anything).Return(nil, expErr)
+	resp, err := e.Grant(context.Background(), 1)
+	require.Equal(t, expErr, err)
+	require.Nil(t, resp)
+}
+
+func TestKeepAliveOnce(t *testing.T) {
+	e := NewMockedETCD(t)
+	expErr := fmt.Errorf("exp")
+	e.cliv3.(*mocks.ETCDClientV3).On("KeepAliveOnce", mock.Anything, mock.Anything).Return(nil, expErr)
+	resp, err := e.KeepAliveOnce(context.Background(), 1)
+	require.Equal(t, expErr, err)
+	require.Nil(t, resp)
+}
+
+func TestTxn(t *testing.T) {
+	e := NewMockedETCD(t)
+	expTxn := &mocks.Txn{}
+	e.cliv3.(*mocks.ETCDClientV3).On("Txn", mock.Anything).Return(expTxn)
+	require.Equal(t, expTxn, e.Txn(context.Background()))
+}
+
 func NewMockedETCD(t *testing.T) *ETCD {
 	e := NewEmbeddedETCD(t)
 	e.cliv3 = &mocks.ETCDClientV3{}

store/etcdv3/meta/meta.go

@@ -12,7 +12,9 @@ import (
 
 // KV .
 type KV interface {
-	ClientV3() *clientv3.Client
+	Grant(ctx context.Context, ttl int64) (*clientv3.LeaseGrantResponse, error)
+	KeepAliveOnce(ctx context.Context, id clientv3.LeaseID) (*clientv3.LeaseKeepAliveResponse, error)
+	Txn(context.Context) clientv3.Txn
 
 	Get(ctx context.Context, key string, opts ...clientv3.OpOption) (*clientv3.GetResponse, error)
 	GetOne(ctx context.Context, key string, opts ...clientv3.OpOption) (*mvccpb.KeyValue, error)