proto reflect

Author: jason-ksztone

workflow/utils.go

@@ -124,6 +124,26 @@ func (wf *Workflow) stepResultToGrpc(s *stepResult, reply interface{}) error {
 	return s.Error
 }
 
+func (wf *Workflow) stepResultFromKitex(reply interface{}, err error) *stepResult {
+	sr := &stepResult{Error: wf.Options.GRPCError2DtmError(err)}
+	if err != nil {
+		sr.Status = dtmcli.StatusFailed
+	}
+	if sr.Error == nil {
+		sr.Data = dtmimp.MustMarshal(reply)
+	} else if sr.Status == dtmcli.StatusFailed {
+		sr.Data = []byte(err.Error())
+	}
+	return sr
+}
+
+func (wf *Workflow) stepResultToKitex(s *stepResult, reply interface{}) error {
+	if s.Error == nil && s.Status == dtmcli.StatusSucceed {
+		dtmimp.MustUnmarshal(s.Data, &reply)
+	}
+	return s.Error
+}
+
 func (wf *Workflow) stepResultFromHTTP(resp *http.Response, err error) *stepResult {
 	sr := &stepResult{Error: err}
 	if err == nil {

workflow/workflow.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"github.com/cloudwego/kitex/pkg/endpoint"
 	"github.com/cloudwego/kitex/pkg/rpcinfo"
+	"google.golang.org/protobuf/reflect/protoreflect"
 	"net/http"
 	"net/url"
 
@@ -264,6 +265,10 @@ func Interceptor(ctx context.Context, method string, req, reply interface{}, cc
 	return wf.stepResultToGrpc(sr, reply)
 }
 
+type KitexProtoReflect interface {
+	ProtoReflect() protoreflect.Message
+}
+
 // KitexInterceptor is the middleware for workflow to capture grpc call result
 func KitexInterceptor(next endpoint.Endpoint) endpoint.Endpoint {
 	return func(ctx context.Context, req, resp interface{}) (err error) {
@@ -292,9 +297,9 @@ func KitexInterceptor(next endpoint.Endpoint) endpoint.Endpoint {
 		}
 		sr := wf.recordedDo(func(bb *dtmcli.BranchBarrier) *stepResult {
 			err := origin()
-			return wf.stepResultFromGrpc(resp, err)
+			return wf.stepResultFromKitex(resp, err)
 		})
-		return wf.stepResultToGrpc(sr, resp)
+		return wf.stepResultToKitex(sr, resp)
 	}
 
 }