feat: move proto related code to a different package

pkg/grpc/dapr/dapr_api_state.go

@@ -0,0 +1,129 @@
+/*
+ * Copyright 2021 Layotto Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package dapr
+
+import (
+	"context"
+	"github.com/dapr/components-contrib/state"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/emptypb"
+	dapr_common_v1pb "mosn.io/layotto/pkg/grpc/dapr/proto/common/v1"
+	dapr_v1pb "mosn.io/layotto/pkg/grpc/dapr/proto/runtime/v1"
+	"mosn.io/layotto/pkg/messages"
+	state2 "mosn.io/layotto/pkg/runtime/state"
+	"mosn.io/pkg/log"
+)
+
+func (d *daprGrpcAPI) SaveState(ctx context.Context, in *dapr_v1pb.SaveStateRequest) (*emptypb.Empty, error) {
+	if in == nil {
+		return &emptypb.Empty{}, status.Error(codes.InvalidArgument, "SaveStateRequest is nil")
+	}
+	// 1. get store
+	store, err := d.getStateStore(in.StoreName)
+	if err != nil {
+		log.DefaultLogger.Errorf("[runtime] [grpc.SaveState] error: %v", err)
+		return &emptypb.Empty{}, err
+	}
+	// 2. convert requests
+	reqs := []state.SetRequest{}
+	for _, s := range in.States {
+		key, err := state2.GetModifiedStateKey(s.Key, in.StoreName, d.appId)
+		if err != nil {
+			return &emptypb.Empty{}, err
+		}
+		reqs = append(reqs, *StateItem2SetRequest(s, key))
+	}
+	// 3. query
+	err = store.BulkSet(reqs)
+	// 4. check result
+	if err != nil {
+		err = d.wrapDaprComponentError(err, messages.ErrStateSave, in.StoreName, err.Error())
+		log.DefaultLogger.Errorf("[runtime] [grpc.SaveState] error: %v", err)
+		return &emptypb.Empty{}, err
+	}
+	return &emptypb.Empty{}, nil
+}
+
+func (d *daprGrpcAPI) getStateStore(name string) (state.Store, error) {
+	if d.stateStores == nil || len(d.stateStores) == 0 {
+		return nil, status.Error(codes.FailedPrecondition, messages.ErrStateStoresNotConfigured)
+	}
+
+	if d.stateStores[name] == nil {
+		return nil, status.Errorf(codes.InvalidArgument, messages.ErrStateStoreNotFound, name)
+	}
+	return d.stateStores[name], nil
+}
+
+func StateItem2SetRequest(grpcReq *dapr_common_v1pb.StateItem, key string) *state.SetRequest {
+	req := &state.SetRequest{
+		Key: key,
+	}
+	if grpcReq == nil {
+		return req
+	}
+	req.Metadata = grpcReq.Metadata
+	req.Value = grpcReq.Value
+	if grpcReq.Etag != nil {
+		req.ETag = &grpcReq.Etag.Value
+	}
+	if grpcReq.Options != nil {
+		req.Options = state.SetStateOption{
+			Consistency: StateConsistencyToString(grpcReq.Options.Consistency),
+			Concurrency: StateConcurrencyToString(grpcReq.Options.Concurrency),
+		}
+	}
+	return req
+}
+
+func StateConsistencyToString(c dapr_common_v1pb.StateOptions_StateConsistency) string {
+	switch c {
+	case dapr_common_v1pb.StateOptions_CONSISTENCY_EVENTUAL:
+		return "eventual"
+	case dapr_common_v1pb.StateOptions_CONSISTENCY_STRONG:
+		return "strong"
+	}
+	return ""
+}
+
+func StateConcurrencyToString(c dapr_common_v1pb.StateOptions_StateConcurrency) string {
+	switch c {
+	case dapr_common_v1pb.StateOptions_CONCURRENCY_FIRST_WRITE:
+		return "first-write"
+	case dapr_common_v1pb.StateOptions_CONCURRENCY_LAST_WRITE:
+		return "last-write"
+	}
+
+	return ""
+}
+
+// wrapDaprComponentError parse and wrap error from dapr component
+func (d *daprGrpcAPI) wrapDaprComponentError(err error, format string, args ...interface{}) error {
+	e, ok := err.(*state.ETagError)
+	if !ok {
+		return status.Errorf(codes.Internal, format, args...)
+	}
+	switch e.Kind() {
+	case state.ETagMismatch:
+		return status.Errorf(codes.Aborted, format, args...)
+	case state.ETagInvalid:
+		return status.Errorf(codes.InvalidArgument, format, args...)
+	}
+
+	return status.Errorf(codes.Internal, format, args...)
+}

pkg/grpc/dapr/dapr_api_unimplement.go

@@ -30,10 +30,6 @@ func (d *daprGrpcAPI) GetBulkState(ctx context.Context, request *runtime.GetBulk
 	panic("implement me")
 }
 
-func (d *daprGrpcAPI) SaveState(ctx context.Context, request *runtime.SaveStateRequest) (*emptypb.Empty, error) {
-	panic("implement me")
-}
-
 func (d *daprGrpcAPI) QueryStateAlpha1(ctx context.Context, request *runtime.QueryStateRequest) (*runtime.QueryStateResponse, error) {
 	panic("implement me")
 }

pkg/grpc/default_api/api.go

@@ -25,6 +25,7 @@ import (
 	"mosn.io/layotto/pkg/grpc/dapr"
 	dapr_common_v1pb "mosn.io/layotto/pkg/grpc/dapr/proto/common/v1"
 	dapr_v1pb "mosn.io/layotto/pkg/grpc/dapr/proto/runtime/v1"
+	state2 "mosn.io/layotto/pkg/runtime/state"
 	mgrpc "mosn.io/mosn/pkg/filter/network/grpc"
 	"strings"
 	"sync"
@@ -42,7 +43,6 @@ import (
 	"github.com/golang/protobuf/ptypes/empty"
 	"mosn.io/layotto/components/file"
 
-	"mosn.io/layotto/pkg/converter"
 	runtime_lock "mosn.io/layotto/pkg/runtime/lock"
 	runtime_sequencer "mosn.io/layotto/pkg/runtime/sequencer"
 
@@ -62,7 +62,6 @@ import (
 	"mosn.io/layotto/components/rpc"
 	"mosn.io/layotto/components/sequencer"
 	"mosn.io/layotto/pkg/messages"
-	runtime_state "mosn.io/layotto/pkg/runtime/state"
 	runtimev1pb "mosn.io/layotto/spec/proto/runtime/v1"
 	"mosn.io/pkg/log"
 )
@@ -494,15 +493,15 @@ func (a *api) GetState(ctx context.Context, in *runtimev1pb.GetStateRequest) (*r
 		return nil, err
 	}
 	// 2. generate the actual key
-	key, err := runtime_state.GetModifiedStateKey(in.Key, in.StoreName, a.appId)
+	key, err := state2.GetModifiedStateKey(in.Key, in.StoreName, a.appId)
 	if err != nil {
 		return &runtimev1pb.GetStateResponse{}, err
 	}
 	req := state.GetRequest{
 		Key:      key,
 		Metadata: in.Metadata,
 		Options: state.GetStateOption{
-			Consistency: runtime_state.StateConsistencyToString(in.Consistency),
+			Consistency: StateConsistencyToString(in.Consistency),
 		},
 	}
 	// 3. query
@@ -514,7 +513,7 @@ func (a *api) GetState(ctx context.Context, in *runtimev1pb.GetStateRequest) (*r
 		return &runtimev1pb.GetStateResponse{}, err
 	}
 
-	return converter.GetResponse2GetStateResponse(compResp), nil
+	return GetResponse2GetStateResponse(compResp), nil
 }
 
 func (a *api) getStateStore(name string) (state.Store, error) {
@@ -545,7 +544,7 @@ func (a *api) GetBulkState(ctx context.Context, in *runtimev1pb.GetBulkStateRequ
 	// 2.1. convert reqs
 	reqs := make([]state.GetRequest, len(in.Keys))
 	for i, k := range in.Keys {
-		key, err := runtime_state.GetModifiedStateKey(k, in.StoreName, a.appId)
+		key, err := state2.GetModifiedStateKey(k, in.StoreName, a.appId)
 		if err != nil {
 			return &runtimev1pb.GetBulkStateResponse{}, err
 		}
@@ -563,7 +562,7 @@ func (a *api) GetBulkState(ctx context.Context, in *runtimev1pb.GetBulkStateRequ
 	// 2.3. parse and return result if store supports this method
 	if support {
 		for i := 0; i < len(responses); i++ {
-			bulkResp.Items = append(bulkResp.Items, converter.BulkGetResponse2BulkStateItem(&responses[i]))
+			bulkResp.Items = append(bulkResp.Items, BulkGetResponse2BulkStateItem(&responses[i]))
 		}
 		return bulkResp, nil
 	}
@@ -597,11 +596,11 @@ func generateGetStateTask(store state.Store, req *state.GetRequest, resultCh cha
 		var item *runtimev1pb.BulkStateItem
 		if err != nil {
 			item = &runtimev1pb.BulkStateItem{
-				Key:   runtime_state.GetOriginalStateKey(req.Key),
+				Key:   state2.GetOriginalStateKey(req.Key),
 				Error: err.Error(),
 			}
 		} else {
-			item = converter.GetResponse2BulkStateItem(r, runtime_state.GetOriginalStateKey(req.Key))
+			item = GetResponse2BulkStateItem(r, state2.GetOriginalStateKey(req.Key))
 		}
 		// collect result
 		select {
@@ -613,33 +612,6 @@ func generateGetStateTask(store state.Store, req *state.GetRequest, resultCh cha
 	}
 }
 
-func (a *api) SaveState(ctx context.Context, in *runtimev1pb.SaveStateRequest) (*emptypb.Empty, error) {
-	// 1. get store
-	store, err := a.getStateStore(in.StoreName)
-	if err != nil {
-		log.DefaultLogger.Errorf("[runtime] [grpc.SaveState] error: %v", err)
-		return &emptypb.Empty{}, err
-	}
-	// 2. convert requests
-	reqs := []state.SetRequest{}
-	for _, s := range in.States {
-		key, err := runtime_state.GetModifiedStateKey(s.Key, in.StoreName, a.appId)
-		if err != nil {
-			return &emptypb.Empty{}, err
-		}
-		reqs = append(reqs, *converter.StateItem2SetRequest(s, key))
-	}
-	// 3. query
-	err = store.BulkSet(reqs)
-	// 4. check result
-	if err != nil {
-		err = a.wrapDaprComponentError(err, messages.ErrStateSave, in.StoreName, err.Error())
-		log.DefaultLogger.Errorf("[runtime] [grpc.SaveState] error: %v", err)
-		return &emptypb.Empty{}, err
-	}
-	return &emptypb.Empty{}, nil
-}
-
 // wrapDaprComponentError parse and wrap error from dapr component
 func (a *api) wrapDaprComponentError(err error, format string, args ...interface{}) error {
 	e, ok := err.(*state.ETagError)
@@ -664,12 +636,12 @@ func (a *api) DeleteState(ctx context.Context, in *runtimev1pb.DeleteStateReques
 		return &emptypb.Empty{}, err
 	}
 	// 2. generate the actual key
-	key, err := runtime_state.GetModifiedStateKey(in.Key, in.StoreName, a.appId)
+	key, err := state2.GetModifiedStateKey(in.Key, in.StoreName, a.appId)
 	if err != nil {
 		return &empty.Empty{}, err
 	}
 	// 3. convert and send request
-	err = store.Delete(converter.DeleteStateRequest2DeleteRequest(in, key))
+	err = store.Delete(DeleteStateRequest2DeleteRequest(in, key))
 	// 4. check result
 	if err != nil {
 		err = a.wrapDaprComponentError(err, messages.ErrStateDelete, in.Key, err.Error())
@@ -689,11 +661,11 @@ func (a *api) DeleteBulkState(ctx context.Context, in *runtimev1pb.DeleteBulkSta
 	// 2. convert request
 	reqs := make([]state.DeleteRequest, 0, len(in.States))
 	for _, item := range in.States {
-		key, err := runtime_state.GetModifiedStateKey(item.Key, in.StoreName, a.appId)
+		key, err := state2.GetModifiedStateKey(item.Key, in.StoreName, a.appId)
 		if err != nil {
 			return &empty.Empty{}, err
 		}
-		reqs = append(reqs, *converter.StateItem2DeleteRequest(item, key))
+		reqs = append(reqs, *StateItem2DeleteRequest(item, key))
 	}
 	// 3. send request
 	err = store.BulkDelete(reqs)
@@ -736,7 +708,7 @@ func (a *api) ExecuteStateTransaction(ctx context.Context, in *runtimev1pb.Execu
 			log.DefaultLogger.Warnf("[runtime] [grpc.ExecuteStateTransaction] one of TransactionalStateOperation.Request is nil")
 			continue
 		}
-		key, err := runtime_state.GetModifiedStateKey(req.Key, in.StoreName, a.appId)
+		key, err := state2.GetModifiedStateKey(req.Key, in.StoreName, a.appId)
 		if err != nil {
 			return &emptypb.Empty{}, err
 		}
@@ -745,12 +717,12 @@ func (a *api) ExecuteStateTransaction(ctx context.Context, in *runtimev1pb.Execu
 		case state.Upsert:
 			operation = state.TransactionalStateOperation{
 				Operation: state.Upsert,
-				Request:   *converter.StateItem2SetRequest(req, key),
+				Request:   *StateItem2SetRequest(req, key),
 			}
 		case state.Delete:
 			operation = state.TransactionalStateOperation{
 				Operation: state.Delete,
-				Request:   *converter.StateItem2DeleteRequest(req, key),
+				Request:   *StateItem2DeleteRequest(req, key),
 			}
 		default:
 			err := status.Errorf(codes.Unimplemented, messages.ErrNotSupportedStateOperation, op.OperationType)
@@ -975,7 +947,7 @@ func (a *api) TryLock(ctx context.Context, req *runtimev1pb.TryLockRequest) (*ru
 		return &runtimev1pb.TryLockResponse{}, status.Errorf(codes.InvalidArgument, messages.ErrLockStoreNotFound, req.StoreName)
 	}
 	// 3. convert request
-	compReq := converter.TryLockRequest2ComponentRequest(req)
+	compReq := TryLockRequest2ComponentRequest(req)
 	// modify key
 	var err error
 	compReq.ResourceId, err = runtime_lock.GetModifiedLockKey(compReq.ResourceId, req.StoreName, a.appId)
@@ -990,7 +962,7 @@ func (a *api) TryLock(ctx context.Context, req *runtimev1pb.TryLockRequest) (*ru
 		return &runtimev1pb.TryLockResponse{}, err
 	}
 	// 5. convert response
-	resp := converter.TryLockResponse2GrpcResponse(compResp)
+	resp := TryLockResponse2GrpcResponse(compResp)
 	return resp, nil
 }
 
@@ -1015,7 +987,7 @@ func (a *api) Unlock(ctx context.Context, req *runtimev1pb.UnlockRequest) (*runt
 		return newInternalErrorUnlockResponse(), status.Errorf(codes.InvalidArgument, messages.ErrLockStoreNotFound, req.StoreName)
 	}
 	// 3. convert request
-	compReq := converter.UnlockGrpc2ComponentRequest(req)
+	compReq := UnlockGrpc2ComponentRequest(req)
 	// modify key
 	var err error
 	compReq.ResourceId, err = runtime_lock.GetModifiedLockKey(compReq.ResourceId, req.StoreName, a.appId)
@@ -1030,7 +1002,7 @@ func (a *api) Unlock(ctx context.Context, req *runtimev1pb.UnlockRequest) (*runt
 		return newInternalErrorUnlockResponse(), err
 	}
 	// 5. convert response
-	resp := converter.UnlockComp2GrpcResponse(compResp)
+	resp := UnlockComp2GrpcResponse(compResp)
 	return resp, nil
 }
 
@@ -1052,7 +1024,7 @@ func (a *api) GetNextId(ctx context.Context, req *runtimev1pb.GetNextIdRequest)
 		return &runtimev1pb.GetNextIdResponse{}, err
 	}
 	// 2. convert
-	compReq, err := converter.GetNextIdRequest2ComponentRequest(req)
+	compReq, err := GetNextIdRequest2ComponentRequest(req)
 	if err != nil {
 		return &runtimev1pb.GetNextIdResponse{}, err
 	}

pkg/grpc/default_api/api_callback.go

@@ -25,7 +25,6 @@ import (
 	"github.com/dapr/components-contrib/pubsub"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
-	runtime_pubsub "mosn.io/layotto/pkg/runtime/pubsub"
 	runtimev1pb "mosn.io/layotto/spec/proto/runtime/v1"
 	_ "net/http/pprof"
 
@@ -105,7 +104,7 @@ func (a *api) getInterestedTopics() (map[string]TopicSubscriptions, error) {
 
 	// 2. handle app subscriptions
 	client := runtimev1pb.NewAppCallbackClient(a.AppCallbackConn)
-	subscriptions = runtime_pubsub.ListTopicSubscriptions(client, log.DefaultLogger)
+	subscriptions = ListTopicSubscriptions(client, log.DefaultLogger)
 	// TODO handle declarative subscriptions
 
 	// 3. prepare result