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

Author: skyao

client/client.go

@@ -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
 

client/client_test.go

@@ -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()

client/state.go

@@ -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")

client/state_test.go

@@ -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()