add bulk delete state method in client sdk (#139)

This commit is contained in:
Sky/敖小剑 2021-02-10 05:55:34 +08:00 committed by GitHub
parent 8114d3f1ab
commit 22f00ec23f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 174 additions and 0 deletions

View File

@ -84,6 +84,12 @@ type Client interface {
// ExecuteStateTransaction provides way to execute multiple operations on a specified store.
ExecuteStateTransaction(ctx context.Context, storeName string, meta map[string]string, ops []*StateOperation) error
// DeleteBulkState deletes content for multiple keys from store.
DeleteBulkState(ctx context.Context, storeName string, keys []string) error
// DeleteBulkState deletes content for multiple keys from store.
DeleteBulkStateItems(ctx context.Context, storeName string, items []*DeleteStateItem) error
// WithTraceID adds existing trace ID to the outgoing context.
WithTraceID(ctx context.Context, id string) context.Context

View File

@ -150,6 +150,13 @@ func (s *testDaprServer) DeleteState(ctx context.Context, req *pb.DeleteStateReq
return &empty.Empty{}, nil
}
func (s *testDaprServer) DeleteBulkState(ctx context.Context, req *pb.DeleteBulkStateRequest) (*empty.Empty, error) {
for _, item := range req.States {
delete(s.state, item.Key)
}
return &empty.Empty{}, nil
}
func (s *testDaprServer) ExecuteStateTransaction(ctx context.Context, in *pb.ExecuteStateTransactionRequest) (*empty.Empty, error) {
for _, op := range in.GetOperations() {
item := op.GetRequest()

View File

@ -125,6 +125,9 @@ type SetStateItem struct {
Options *StateOptions
}
// DeleteStateItem represents a single state to be deleted.
type DeleteStateItem SetStateItem
// ETag represents an versioned record information
type ETag struct {
Value string
@ -339,6 +342,58 @@ func (c *GRPCClient) DeleteStateWithETag(ctx context.Context, storeName, key str
return nil
}
// DeleteBulkState deletes content for multiple keys from store.
func (c *GRPCClient) DeleteBulkState(ctx context.Context, storeName string, keys []string) error {
if len(keys) == 0 {
return nil
}
items := make([]*DeleteStateItem, 0, len(keys))
for i := 0; i < len(keys); i++ {
item := &DeleteStateItem{
Key: keys[i],
}
items = append(items, item)
}
return c.DeleteBulkStateItems(ctx, storeName, items)
}
// DeleteBulkState deletes content for multiple keys from store.
func (c *GRPCClient) DeleteBulkStateItems(ctx context.Context, storeName string, items []*DeleteStateItem) error {
if len(items) == 0 {
return nil
}
states := make([]*v1.StateItem, 0, len(items))
for i := 0; i < len(items); i++ {
item := items[i]
if err := hasRequiredStateArgs(storeName, item.Key); err != nil {
return errors.Wrap(err, "missing required arguments")
}
state := &v1.StateItem{
Key: item.Key,
Metadata: item.Metadata,
Options: toProtoStateOptions(item.Options),
}
if item.Etag != nil {
state.Etag = &v1.Etag{
Value: item.Etag.Value,
}
}
states = append(states, state)
}
req := &pb.DeleteBulkStateRequest{
StoreName: storeName,
States: states,
}
_, err := c.protoClient.DeleteBulkState(c.withAuthToken(ctx), req)
return err
}
func hasRequiredStateArgs(storeName, key string) error {
if storeName == "" {
return errors.New("store")

View File

@ -163,6 +163,112 @@ func TestDeleteState(t *testing.T) {
})
}
func TestDeleteBulkState(t *testing.T) {
ctx := context.Background()
data := "test"
store := "test"
keys := []string{"key1", "key2", "key3"}
t.Run("delete not exist data", func(t *testing.T) {
err := testClient.DeleteBulkState(ctx, store, keys)
assert.Nil(t, err)
})
t.Run("delete not exist data with stateIem", func(t *testing.T) {
items := make([]*DeleteStateItem, 0, len(keys))
for _, key := range keys {
items = append(items, &DeleteStateItem{
Key: key,
Metadata: map[string]string{},
Options: &StateOptions{
Concurrency: StateConcurrencyFirstWrite,
Consistency: StateConsistencyEventual,
},
})
}
err := testClient.DeleteBulkStateItems(ctx, store, items)
assert.Nil(t, err)
})
t.Run("delete exist data", func(t *testing.T) {
// save data
items := make([]*SetStateItem, 0, len(keys))
for _, key := range keys {
items = append(items, &SetStateItem{
Key: key,
Value: []byte(data),
Metadata: map[string]string{},
Etag: &ETag{Value: "1"},
Options: &StateOptions{
Concurrency: StateConcurrencyFirstWrite,
Consistency: StateConsistencyEventual,
},
})
}
err := testClient.SaveBulkState(ctx, store, items...)
assert.Nil(t, err)
// confirm data saved
getItems, err := testClient.GetBulkState(ctx, store, keys, nil, 1)
assert.NoError(t, err)
assert.Equal(t, len(keys), len(getItems))
// delete
err = testClient.DeleteBulkState(ctx, store, keys)
assert.NoError(t, err)
// confirm data deleted
getItems, err = testClient.GetBulkState(ctx, store, keys, nil, 1)
assert.NoError(t, err)
assert.Equal(t, 0, len(getItems))
})
t.Run("delete exist data with stateItem", func(t *testing.T) {
// save data
items := make([]*SetStateItem, 0, len(keys))
for _, key := range keys {
items = append(items, &SetStateItem{
Key: key,
Value: []byte(data),
Metadata: map[string]string{},
Etag: &ETag{Value: "1"},
Options: &StateOptions{
Concurrency: StateConcurrencyFirstWrite,
Consistency: StateConsistencyEventual,
},
})
}
err := testClient.SaveBulkState(ctx, store, items...)
assert.Nil(t, err)
// confirm data saved
getItems, err := testClient.GetBulkState(ctx, store, keys, nil, 1)
assert.NoError(t, err)
assert.Equal(t, len(keys), len(getItems))
// delete
deleteItems := make([]*DeleteStateItem, 0, len(keys))
for _, key := range keys {
deleteItems = append(deleteItems, &DeleteStateItem{
Key: key,
Metadata: map[string]string{},
Etag: &ETag{Value: "1"},
Options: &StateOptions{
Concurrency: StateConcurrencyFirstWrite,
Consistency: StateConsistencyEventual,
},
})
}
err = testClient.DeleteBulkStateItems(ctx, store, deleteItems)
assert.Nil(t, err)
// confirm data deleted
getItems, err = testClient.GetBulkState(ctx, store, keys, nil, 1)
assert.NoError(t, err)
assert.Equal(t, 0, len(getItems))
})
}
// go test -timeout 30s ./client -count 1 -run ^TestStateTransactions$
func TestStateTransactions(t *testing.T) {
ctx := context.Background()