mirror of https://github.com/dtm-labs/dtm.git
yedf2
4 years ago
47 changed files with 1888 additions and 246 deletions
@ -0,0 +1,60 @@ |
|||
package dtmcli |
|||
|
|||
import ( |
|||
"errors" |
|||
"fmt" |
|||
"net/http" |
|||
|
|||
"github.com/dtm-labs/dtm/dtmcli/dtmimp" |
|||
"github.com/go-resty/resty/v2" |
|||
) |
|||
|
|||
// MustGenGid generate a new gid
|
|||
func MustGenGid(server string) string { |
|||
res := map[string]string{} |
|||
resp, err := dtmimp.RestyClient.R().SetResult(&res).Get(server + "/newGid") |
|||
if err != nil || res["gid"] == "" { |
|||
panic(fmt.Errorf("newGid error: %v, resp: %s", err, resp)) |
|||
} |
|||
return res["gid"] |
|||
} |
|||
|
|||
// String2DtmError translate string to dtm error
|
|||
func String2DtmError(str string) error { |
|||
return map[string]error{ |
|||
ResultFailure: ErrFailure, |
|||
ResultOngoing: ErrOngoing, |
|||
ResultSuccess: nil, |
|||
"": nil, |
|||
}[str] |
|||
} |
|||
|
|||
// Result2HttpJSON return the http code and json result
|
|||
// if result is error, the return proper code, else return StatusOK
|
|||
func Result2HttpJSON(result interface{}) (code int, res interface{}) { |
|||
err, _ := result.(error) |
|||
if err == nil { |
|||
code = http.StatusOK |
|||
res = result |
|||
} else { |
|||
res = map[string]string{ |
|||
"error": err.Error(), |
|||
} |
|||
if errors.Is(err, ErrFailure) { |
|||
code = http.StatusConflict |
|||
} else if errors.Is(err, ErrOngoing) { |
|||
code = http.StatusTooEarly |
|||
} else if err != nil { |
|||
code = http.StatusInternalServerError |
|||
} |
|||
} |
|||
return |
|||
} |
|||
|
|||
func IsRollback(resp *resty.Response, err error) bool { |
|||
return err == ErrFailure || dtmimp.RespAsErrorCompatible(resp) == ErrFailure |
|||
} |
|||
|
|||
func IsOngoing(resp *resty.Response, err error) bool { |
|||
return err == ErrOngoing || dtmimp.RespAsErrorCompatible(resp) == ErrOngoing |
|||
} |
|||
@ -0,0 +1,53 @@ |
|||
package workflow |
|||
|
|||
import ( |
|||
"bytes" |
|||
"io" |
|||
"strings" |
|||
) |
|||
|
|||
// NewRespBodyFromString creates an io.ReadCloser from a string that
|
|||
// is suitable for use as an http response body.
|
|||
//
|
|||
// To pass the content of an existing file as body use httpmock.File as in:
|
|||
// httpmock.NewRespBodyFromString(httpmock.File("body.txt").String())
|
|||
func NewRespBodyFromString(body string) io.ReadCloser { |
|||
return &dummyReadCloser{orig: body} |
|||
} |
|||
|
|||
// NewRespBodyFromBytes creates an io.ReadCloser from a byte slice
|
|||
// that is suitable for use as an http response body.
|
|||
//
|
|||
// To pass the content of an existing file as body use httpmock.File as in:
|
|||
// httpmock.NewRespBodyFromBytes(httpmock.File("body.txt").Bytes())
|
|||
func NewRespBodyFromBytes(body []byte) io.ReadCloser { |
|||
return &dummyReadCloser{orig: body} |
|||
} |
|||
|
|||
type dummyReadCloser struct { |
|||
orig interface{} // string or []byte
|
|||
body io.ReadSeeker // instanciated on demand from orig
|
|||
} |
|||
|
|||
// setup ensures d.body is correctly initialized.
|
|||
func (d *dummyReadCloser) setup() { |
|||
if d.body == nil { |
|||
switch body := d.orig.(type) { |
|||
case string: |
|||
d.body = strings.NewReader(body) |
|||
case []byte: |
|||
d.body = bytes.NewReader(body) |
|||
} |
|||
} |
|||
} |
|||
|
|||
func (d *dummyReadCloser) Read(p []byte) (n int, err error) { |
|||
d.setup() |
|||
return d.body.Read(p) |
|||
} |
|||
|
|||
func (d *dummyReadCloser) Close() error { |
|||
d.setup() |
|||
d.body.Seek(0, io.SeekEnd) // nolint: errcheck
|
|||
return nil |
|||
} |
|||
@ -0,0 +1,46 @@ |
|||
package workflow |
|||
|
|||
import ( |
|||
"fmt" |
|||
"net/url" |
|||
|
|||
"github.com/dtm-labs/dtm/dtmcli/logger" |
|||
) |
|||
|
|||
type workflowFactory struct { |
|||
protocol string |
|||
httpDtm string |
|||
httpCallback string |
|||
grpcDtm string |
|||
grpcCallback string |
|||
handlers map[string]WfFunc |
|||
} |
|||
|
|||
var defaultFac = workflowFactory{ |
|||
handlers: map[string]WfFunc{}, |
|||
} |
|||
|
|||
func (w *workflowFactory) execute(name string, gid string, data []byte) error { |
|||
handler := w.handlers[name] |
|||
if handler == nil { |
|||
return fmt.Errorf("workflow '%s' not registered. please register at startup", name) |
|||
} |
|||
wf := w.newWorkflow(name, gid, data) |
|||
return wf.process(handler, data) |
|||
} |
|||
|
|||
func (w *workflowFactory) executeByQS(qs url.Values, body []byte) error { |
|||
name := qs.Get("op") |
|||
gid := qs.Get("gid") |
|||
return w.execute(name, gid, body) |
|||
} |
|||
|
|||
func (w *workflowFactory) register(name string, handler WfFunc) error { |
|||
e := w.handlers[name] |
|||
if e != nil { |
|||
return fmt.Errorf("a handler already exists for %s", name) |
|||
} |
|||
logger.Debugf("workflow '%s' registered.", name) |
|||
w.handlers[name] = handler |
|||
return nil |
|||
} |
|||
@ -0,0 +1,191 @@ |
|||
package workflow |
|||
|
|||
import ( |
|||
"context" |
|||
"errors" |
|||
|
|||
"github.com/dtm-labs/dtm/dtmcli" |
|||
"github.com/dtm-labs/dtm/dtmcli/dtmimp" |
|||
"github.com/dtm-labs/dtm/dtmcli/logger" |
|||
"github.com/dtm-labs/dtm/dtmgrpc" |
|||
"github.com/go-resty/resty/v2" |
|||
) |
|||
|
|||
type workflowImp struct { |
|||
restyClient *resty.Client |
|||
idGen dtmimp.BranchIDGen |
|||
currentBranch string |
|||
progresses map[string]*stepResult |
|||
currentOp string |
|||
succeededOps []workflowPhase2Item |
|||
failedOps []workflowPhase2Item |
|||
} |
|||
|
|||
type workflowPhase2Item struct { |
|||
branchID, op string |
|||
fn WfPhase2Func |
|||
} |
|||
|
|||
func (wf *Workflow) loadProgresses() error { |
|||
progresses, err := wf.getProgress() |
|||
if err == nil { |
|||
wf.progresses = map[string]*stepResult{} |
|||
for _, p := range progresses { |
|||
wf.progresses[p.BranchID+"-"+p.Op] = &stepResult{ |
|||
Status: p.Status, |
|||
Data: p.BinData, |
|||
} |
|||
} |
|||
} |
|||
return err |
|||
} |
|||
|
|||
type wfMeta struct{} |
|||
|
|||
func (w *workflowFactory) newWorkflow(name string, gid string, data []byte) *Workflow { |
|||
wf := &Workflow{ |
|||
TransBase: dtmimp.NewTransBase(gid, "workflow", "not inited", ""), |
|||
Name: name, |
|||
workflowImp: workflowImp{ |
|||
idGen: dtmimp.BranchIDGen{}, |
|||
succeededOps: []workflowPhase2Item{}, |
|||
failedOps: []workflowPhase2Item{}, |
|||
currentOp: dtmimp.OpAction, |
|||
}, |
|||
} |
|||
wf.Protocol = w.protocol |
|||
if w.protocol == dtmimp.ProtocolGRPC { |
|||
wf.Dtm = w.grpcDtm |
|||
wf.QueryPrepared = w.grpcCallback |
|||
} else { |
|||
wf.Dtm = w.httpDtm |
|||
wf.QueryPrepared = w.httpCallback |
|||
} |
|||
wf.newBranch() |
|||
wf.CustomData = dtmimp.MustMarshalString(map[string]interface{}{ |
|||
"name": wf.Name, |
|||
"data": data, |
|||
}) |
|||
wf.Context = context.WithValue(wf.Context, wfMeta{}, wf) |
|||
wf.initRestyClient() |
|||
return wf |
|||
} |
|||
|
|||
func (wf *Workflow) initRestyClient() { |
|||
wf.restyClient = resty.New() |
|||
wf.restyClient.OnBeforeRequest(func(c *resty.Client, r *resty.Request) error { |
|||
r.SetQueryParams(map[string]string{ |
|||
"gid": wf.Gid, |
|||
"trans_type": wf.TransType, |
|||
"branch_id": wf.currentBranch, |
|||
"op": dtmimp.OpAction, |
|||
}) |
|||
err := dtmimp.BeforeRequest(c, r) |
|||
return err |
|||
}) |
|||
old := wf.restyClient.GetClient().Transport |
|||
wf.restyClient.GetClient().Transport = NewRoundTripper(old, wf) |
|||
wf.restyClient.OnAfterResponse(func(c *resty.Client, r *resty.Response) error { |
|||
err := dtmimp.AfterResponse(c, r) |
|||
if err == nil && !wf.Options.DisalbeAutoError { |
|||
err = dtmimp.RespAsErrorCompatible(r) // check for dtm error
|
|||
} |
|||
return err |
|||
}) |
|||
} |
|||
|
|||
func (wf *Workflow) process(handler WfFunc, data []byte) (err error) { |
|||
err = wf.prepare() |
|||
if err == nil { |
|||
err = wf.loadProgresses() |
|||
} |
|||
if err == nil { |
|||
err = handler(wf, data) |
|||
err = dtmgrpc.GrpcError2DtmError(err) |
|||
if err != nil && !errors.Is(err, dtmcli.ErrFailure) { |
|||
return err |
|||
} |
|||
err = wf.processPhase2(err) |
|||
} |
|||
if err == nil || errors.Is(err, dtmcli.ErrFailure) { |
|||
err1 := wf.submit(wfErrorToStatus(err)) |
|||
if err1 != nil { |
|||
return err1 |
|||
} |
|||
} |
|||
return err |
|||
|
|||
} |
|||
|
|||
func (wf *Workflow) saveResult(branchID string, op string, sr *stepResult) error { |
|||
if sr.Status == "" { |
|||
return sr.Error |
|||
} |
|||
return wf.registerBranch(sr.Data, branchID, op, sr.Status) |
|||
} |
|||
|
|||
func (wf *Workflow) processPhase2(err error) error { |
|||
ops := wf.succeededOps |
|||
if err == nil { |
|||
wf.currentOp = dtmimp.OpCommit |
|||
} else { |
|||
wf.currentOp = dtmimp.OpRollback |
|||
ops = wf.failedOps |
|||
} |
|||
for i := len(ops) - 1; i >= 0; i-- { |
|||
op := ops[i] |
|||
|
|||
err1 := wf.callPhase2(op.branchID, op.op, op.fn) |
|||
if err1 != nil { |
|||
return err1 |
|||
} |
|||
} |
|||
return err |
|||
} |
|||
|
|||
func (wf *Workflow) callPhase2(branchID string, op string, fn WfPhase2Func) error { |
|||
wf.currentBranch = branchID |
|||
r := wf.recordedDo(func(bb *dtmcli.BranchBarrier) *stepResult { |
|||
err := fn(bb) |
|||
if errors.Is(err, dtmcli.ErrFailure) { |
|||
panic("should not return ErrFail in phase2") |
|||
} |
|||
return stepResultFromLocal(nil, err) |
|||
}) |
|||
_, err := stepResultToLocal(r) |
|||
return err |
|||
} |
|||
|
|||
func (wf *Workflow) recordedDo(fn func(bb *dtmcli.BranchBarrier) *stepResult) *stepResult { |
|||
branchID := wf.currentBranch |
|||
r := wf.getStepResult() |
|||
if wf.currentOp == dtmimp.OpAction { // for action steps, an action will start a new branch
|
|||
wf.newBranch() |
|||
} |
|||
if r != nil { |
|||
logger.Debugf("progress restored: %s %s %v %s %s", branchID, wf.currentOp, r.Error, r.Status, r.Data) |
|||
return r |
|||
} |
|||
bb := &dtmcli.BranchBarrier{ |
|||
TransType: wf.TransType, |
|||
Gid: wf.Gid, |
|||
BranchID: branchID, |
|||
Op: wf.currentOp, |
|||
} |
|||
r = fn(bb) |
|||
err := wf.saveResult(branchID, wf.currentOp, r) |
|||
if err != nil { |
|||
r = stepResultFromLocal(nil, err) |
|||
} |
|||
return r |
|||
} |
|||
|
|||
func (wf *Workflow) newBranch() { |
|||
wf.idGen.NewSubBranchID() |
|||
wf.currentBranch = wf.idGen.CurrentSubBranchID() |
|||
} |
|||
|
|||
func (wf *Workflow) getStepResult() *stepResult { |
|||
logger.Debugf("getStepResult: %s %v", wf.currentBranch+"-"+wf.currentOp, wf.progresses[wf.currentBranch+"-"+wf.currentOp]) |
|||
return wf.progresses[wf.currentBranch+"-"+wf.currentOp] |
|||
} |
|||
@ -0,0 +1,78 @@ |
|||
package workflow |
|||
|
|||
import ( |
|||
"context" |
|||
"strings" |
|||
|
|||
"github.com/dtm-labs/dtm/dtmcli/dtmimp" |
|||
"github.com/dtm-labs/dtm/dtmcli/logger" |
|||
"github.com/dtm-labs/dtm/dtmgrpc/dtmgimp" |
|||
"github.com/dtm-labs/dtm/dtmgrpc/dtmgpb" |
|||
"google.golang.org/protobuf/types/known/emptypb" |
|||
) |
|||
|
|||
func (wf *Workflow) getProgress() ([]*dtmgpb.DtmProgress, error) { |
|||
if wf.Protocol == dtmimp.ProtocolGRPC { |
|||
var reply dtmgpb.DtmProgressesReply |
|||
err := dtmgimp.MustGetGrpcConn(wf.Dtm, false).Invoke(wf.Context, "/dtmgimp.Dtm/Progresses", |
|||
dtmgimp.GetDtmRequest(wf.TransBase), &reply) |
|||
if err == nil { |
|||
return reply.Progresses, nil |
|||
} |
|||
return nil, err |
|||
} |
|||
resp, err := dtmimp.RestyClient.R().SetQueryParam("gid", wf.Gid).Get(wf.Dtm + "/progresses") |
|||
var progresses []*dtmgpb.DtmProgress |
|||
if err == nil { |
|||
dtmimp.MustUnmarshal(resp.Body(), &progresses) |
|||
} |
|||
return progresses, err |
|||
} |
|||
|
|||
func (wf *Workflow) submit(status string) error { |
|||
if wf.Protocol == dtmimp.ProtocolHTTP { |
|||
m := map[string]interface{}{ |
|||
"gid": wf.Gid, |
|||
"trans_type": wf.TransType, |
|||
"req_extra": map[string]string{ |
|||
"status": status, |
|||
}, |
|||
} |
|||
_, err := dtmimp.TransCallDtmExt(wf.TransBase, m, "submit") |
|||
return err |
|||
} |
|||
req := dtmgimp.GetDtmRequest(wf.TransBase) |
|||
req.ReqExtra = map[string]string{ |
|||
"status": status, |
|||
} |
|||
reply := emptypb.Empty{} |
|||
return dtmgimp.MustGetGrpcConn(wf.Dtm, false).Invoke(wf.Context, "/dtmgimp.Dtm/"+"Submit", req, &reply) |
|||
} |
|||
|
|||
func (wf *Workflow) registerBranch(res []byte, branchID string, op string, status string) error { |
|||
logger.Errorf("registerBranch: %s %s %s", branchID, op, status) |
|||
if wf.Protocol == dtmimp.ProtocolHTTP { |
|||
return dtmimp.TransRegisterBranch(wf.TransBase, map[string]string{ |
|||
"data": string(res), |
|||
"branch_id": branchID, |
|||
"op": op, |
|||
"status": status, |
|||
}, "registerBranch") |
|||
} |
|||
_, err := dtmgimp.MustGetDtmClient(wf.Dtm).RegisterBranch(context.Background(), &dtmgpb.DtmBranchRequest{ |
|||
Gid: wf.Gid, |
|||
TransType: wf.TransType, |
|||
BranchID: branchID, |
|||
BusiPayload: res, |
|||
Data: map[string]string{"status": status, "op": op}, |
|||
}) |
|||
return err |
|||
} |
|||
|
|||
func (wf *Workflow) prepare() error { |
|||
operation := "prepare" |
|||
if wf.Protocol == dtmimp.ProtocolGRPC { |
|||
return dtmgimp.DtmGrpcCall(wf.TransBase, strings.Title(operation)) |
|||
} |
|||
return dtmimp.TransCallDtm(wf.TransBase, operation) |
|||
} |
|||
@ -0,0 +1,26 @@ |
|||
package workflow |
|||
|
|||
import ( |
|||
"context" |
|||
|
|||
"github.com/dtm-labs/dtm/dtmcli/dtmimp" |
|||
"github.com/dtm-labs/dtm/dtmgrpc" |
|||
"github.com/dtm-labs/dtm/dtmgrpc/dtmgimp" |
|||
"github.com/dtm-labs/dtm/dtmgrpc/workflow/wfpb" |
|||
"google.golang.org/grpc/codes" |
|||
"google.golang.org/grpc/status" |
|||
"google.golang.org/protobuf/types/known/emptypb" |
|||
) |
|||
|
|||
type workflowServer struct { |
|||
wfpb.UnimplementedWorkflowServer |
|||
} |
|||
|
|||
func (s *workflowServer) Execute(ctx context.Context, wd *wfpb.WorkflowData) (*emptypb.Empty, error) { |
|||
if defaultFac.protocol != dtmimp.ProtocolGRPC { |
|||
return nil, status.Errorf(codes.Internal, "workflow server not inited. please call workflow.InitGrpc first") |
|||
} |
|||
tb := dtmgimp.TransBaseFromGrpc(ctx) |
|||
err := defaultFac.execute(tb.Op, tb.Gid, wd.Data) |
|||
return &emptypb.Empty{}, dtmgrpc.DtmError2GrpcError(err) |
|||
} |
|||
@ -0,0 +1,138 @@ |
|||
package workflow |
|||
|
|||
import ( |
|||
"errors" |
|||
"fmt" |
|||
"io/ioutil" |
|||
"net/http" |
|||
"strconv" |
|||
|
|||
"github.com/dtm-labs/dtm/dtmcli" |
|||
"github.com/dtm-labs/dtm/dtmcli/dtmimp" |
|||
"github.com/dtm-labs/dtm/dtmgrpc/dtmgimp" |
|||
"google.golang.org/grpc/codes" |
|||
"google.golang.org/grpc/status" |
|||
"google.golang.org/protobuf/reflect/protoreflect" |
|||
) |
|||
|
|||
const HBranchID = "dtm-branch-id" |
|||
const HBranchOp = "dtm-branch-op" |
|||
|
|||
func statusToCode(status string) int { |
|||
if status == "succeed" { |
|||
return 200 |
|||
} |
|||
return 409 |
|||
} |
|||
|
|||
func wfErrorToStatus(err error) string { |
|||
if err == nil { |
|||
return dtmcli.StatusSucceed |
|||
} else if errors.Is(err, dtmcli.ErrFailure) { |
|||
return dtmcli.StatusFailed |
|||
} |
|||
return "" |
|||
} |
|||
|
|||
type stepResult struct { |
|||
Error error // if Error != nil || Status == "", result will not be saved
|
|||
Status string // succeed | failed | ""
|
|||
// if status == succeed, data is the result.
|
|||
// if status == failed, data is the error message
|
|||
Data []byte |
|||
} |
|||
|
|||
type roundTripper struct { |
|||
old http.RoundTripper |
|||
wf *Workflow |
|||
} |
|||
|
|||
func newJSONResponse(status int, result []byte) *http.Response { |
|||
return &http.Response{ |
|||
Status: strconv.Itoa(status), |
|||
StatusCode: status, |
|||
Body: NewRespBodyFromBytes(result), |
|||
Header: http.Header{ |
|||
"Content-Type": []string{"application/json"}, |
|||
}, |
|||
ContentLength: -1, |
|||
} |
|||
} |
|||
|
|||
func (r *roundTripper) RoundTrip(req *http.Request) (*http.Response, error) { |
|||
wf := r.wf |
|||
if wf.currentOp != dtmimp.OpAction { // in phase 2, do not save, because it is saved outer
|
|||
return r.old.RoundTrip(req) |
|||
} |
|||
sr := wf.recordedDo(func(bb *dtmcli.BranchBarrier) *stepResult { |
|||
resp, err := r.old.RoundTrip(req) |
|||
return stepResultFromHttp(resp, err) |
|||
}) |
|||
return stepResultToHttp(sr) |
|||
} |
|||
|
|||
func NewRoundTripper(old http.RoundTripper, wf *Workflow) http.RoundTripper { |
|||
return &roundTripper{old: old, wf: wf} |
|||
} |
|||
|
|||
func stepResultFromLocal(data []byte, err error) *stepResult { |
|||
return &stepResult{ |
|||
Error: err, |
|||
Status: wfErrorToStatus(err), |
|||
Data: data, |
|||
} |
|||
} |
|||
|
|||
func stepResultToLocal(s *stepResult) ([]byte, error) { |
|||
if s.Error != nil { |
|||
return nil, s.Error |
|||
} else if s.Status == dtmcli.StatusFailed { |
|||
return nil, fmt.Errorf("%s. %w", string(s.Data), dtmcli.ErrFailure) |
|||
} |
|||
return s.Data, nil |
|||
} |
|||
|
|||
func stepResultFromGrpc(reply interface{}, err error) *stepResult { |
|||
sr := &stepResult{} |
|||
st, ok := status.FromError(err) |
|||
if err == nil { |
|||
sr.Status = dtmcli.StatusSucceed |
|||
sr.Data = dtmgimp.MustProtoMarshal(reply.(protoreflect.ProtoMessage)) |
|||
} else if ok && st.Code() == codes.Aborted { |
|||
sr.Status = dtmcli.StatusFailed |
|||
sr.Data = []byte(st.Message()) |
|||
} else { |
|||
sr.Error = err |
|||
} |
|||
return sr |
|||
} |
|||
|
|||
func stepResultToGrpc(s *stepResult, reply interface{}) error { |
|||
if s.Error != nil { |
|||
return s.Error |
|||
} else if s.Status == dtmcli.StatusSucceed { |
|||
dtmgimp.MustProtoUnmarshal(s.Data, reply.(protoreflect.ProtoMessage)) |
|||
return nil |
|||
} |
|||
return status.New(codes.Aborted, string(s.Data)).Err() |
|||
} |
|||
|
|||
func stepResultFromHttp(resp *http.Response, err error) *stepResult { |
|||
sr := &stepResult{Error: err} |
|||
if err == nil { |
|||
sr.Data, sr.Error = ioutil.ReadAll(resp.Body) |
|||
if resp.StatusCode == http.StatusOK { |
|||
sr.Status = dtmcli.StatusSucceed |
|||
} else if resp.StatusCode == http.StatusConflict { |
|||
sr.Status = dtmcli.StatusFailed |
|||
} |
|||
} |
|||
return sr |
|||
} |
|||
|
|||
func stepResultToHttp(s *stepResult) (*http.Response, error) { |
|||
if s.Error != nil { |
|||
return nil, s.Error |
|||
} |
|||
return newJSONResponse(statusToCode(s.Status), s.Data), nil |
|||
} |
|||
@ -0,0 +1,153 @@ |
|||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
|||
// versions:
|
|||
// protoc-gen-go v1.28.0
|
|||
// protoc v3.17.3
|
|||
// source: dtmgrpc/workflow/wfpb/wf.proto
|
|||
|
|||
package wfpb |
|||
|
|||
import ( |
|||
protoreflect "google.golang.org/protobuf/reflect/protoreflect" |
|||
protoimpl "google.golang.org/protobuf/runtime/protoimpl" |
|||
emptypb "google.golang.org/protobuf/types/known/emptypb" |
|||
reflect "reflect" |
|||
sync "sync" |
|||
) |
|||
|
|||
const ( |
|||
// Verify that this generated code is sufficiently up-to-date.
|
|||
_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) |
|||
// Verify that runtime/protoimpl is sufficiently up-to-date.
|
|||
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) |
|||
) |
|||
|
|||
type WorkflowData struct { |
|||
state protoimpl.MessageState |
|||
sizeCache protoimpl.SizeCache |
|||
unknownFields protoimpl.UnknownFields |
|||
|
|||
Data []byte `protobuf:"bytes,1,opt,name=Data,proto3" json:"Data,omitempty"` |
|||
} |
|||
|
|||
func (x *WorkflowData) Reset() { |
|||
*x = WorkflowData{} |
|||
if protoimpl.UnsafeEnabled { |
|||
mi := &file_dtmgrpc_workflow_wfpb_wf_proto_msgTypes[0] |
|||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) |
|||
ms.StoreMessageInfo(mi) |
|||
} |
|||
} |
|||
|
|||
func (x *WorkflowData) String() string { |
|||
return protoimpl.X.MessageStringOf(x) |
|||
} |
|||
|
|||
func (*WorkflowData) ProtoMessage() {} |
|||
|
|||
func (x *WorkflowData) ProtoReflect() protoreflect.Message { |
|||
mi := &file_dtmgrpc_workflow_wfpb_wf_proto_msgTypes[0] |
|||
if protoimpl.UnsafeEnabled && x != nil { |
|||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) |
|||
if ms.LoadMessageInfo() == nil { |
|||
ms.StoreMessageInfo(mi) |
|||
} |
|||
return ms |
|||
} |
|||
return mi.MessageOf(x) |
|||
} |
|||
|
|||
// Deprecated: Use WorkflowData.ProtoReflect.Descriptor instead.
|
|||
func (*WorkflowData) Descriptor() ([]byte, []int) { |
|||
return file_dtmgrpc_workflow_wfpb_wf_proto_rawDescGZIP(), []int{0} |
|||
} |
|||
|
|||
func (x *WorkflowData) GetData() []byte { |
|||
if x != nil { |
|||
return x.Data |
|||
} |
|||
return nil |
|||
} |
|||
|
|||
var File_dtmgrpc_workflow_wfpb_wf_proto protoreflect.FileDescriptor |
|||
|
|||
var file_dtmgrpc_workflow_wfpb_wf_proto_rawDesc = []byte{ |
|||
0x0a, 0x1e, 0x64, 0x74, 0x6d, 0x67, 0x72, 0x70, 0x63, 0x2f, 0x77, 0x6f, 0x72, 0x6b, 0x66, 0x6c, |
|||
0x6f, 0x77, 0x2f, 0x77, 0x66, 0x70, 0x62, 0x2f, 0x77, 0x66, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, |
|||
0x12, 0x08, 0x77, 0x6f, 0x72, 0x6b, 0x66, 0x6c, 0x6f, 0x77, 0x1a, 0x1b, 0x67, 0x6f, 0x6f, 0x67, |
|||
0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2f, 0x65, 0x6d, 0x70, 0x74, |
|||
0x79, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x22, 0x0a, 0x0c, 0x57, 0x6f, 0x72, 0x6b, 0x66, |
|||
0x6c, 0x6f, 0x77, 0x44, 0x61, 0x74, 0x61, 0x12, 0x12, 0x0a, 0x04, 0x44, 0x61, 0x74, 0x61, 0x18, |
|||
0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x44, 0x61, 0x74, 0x61, 0x32, 0x47, 0x0a, 0x08, 0x57, |
|||
0x6f, 0x72, 0x6b, 0x66, 0x6c, 0x6f, 0x77, 0x12, 0x3b, 0x0a, 0x07, 0x45, 0x78, 0x65, 0x63, 0x75, |
|||
0x74, 0x65, 0x12, 0x16, 0x2e, 0x77, 0x6f, 0x72, 0x6b, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x57, 0x6f, |
|||
0x72, 0x6b, 0x66, 0x6c, 0x6f, 0x77, 0x44, 0x61, 0x74, 0x61, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, |
|||
0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, |
|||
0x74, 0x79, 0x22, 0x00, 0x42, 0x08, 0x5a, 0x06, 0x2e, 0x2f, 0x77, 0x66, 0x70, 0x62, 0x62, 0x06, |
|||
0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, |
|||
} |
|||
|
|||
var ( |
|||
file_dtmgrpc_workflow_wfpb_wf_proto_rawDescOnce sync.Once |
|||
file_dtmgrpc_workflow_wfpb_wf_proto_rawDescData = file_dtmgrpc_workflow_wfpb_wf_proto_rawDesc |
|||
) |
|||
|
|||
func file_dtmgrpc_workflow_wfpb_wf_proto_rawDescGZIP() []byte { |
|||
file_dtmgrpc_workflow_wfpb_wf_proto_rawDescOnce.Do(func() { |
|||
file_dtmgrpc_workflow_wfpb_wf_proto_rawDescData = protoimpl.X.CompressGZIP(file_dtmgrpc_workflow_wfpb_wf_proto_rawDescData) |
|||
}) |
|||
return file_dtmgrpc_workflow_wfpb_wf_proto_rawDescData |
|||
} |
|||
|
|||
var file_dtmgrpc_workflow_wfpb_wf_proto_msgTypes = make([]protoimpl.MessageInfo, 1) |
|||
var file_dtmgrpc_workflow_wfpb_wf_proto_goTypes = []interface{}{ |
|||
(*WorkflowData)(nil), // 0: workflow.WorkflowData
|
|||
(*emptypb.Empty)(nil), // 1: google.protobuf.Empty
|
|||
} |
|||
var file_dtmgrpc_workflow_wfpb_wf_proto_depIdxs = []int32{ |
|||
0, // 0: workflow.Workflow.Execute:input_type -> workflow.WorkflowData
|
|||
1, // 1: workflow.Workflow.Execute:output_type -> google.protobuf.Empty
|
|||
1, // [1:2] is the sub-list for method output_type
|
|||
0, // [0:1] is the sub-list for method input_type
|
|||
0, // [0:0] is the sub-list for extension type_name
|
|||
0, // [0:0] is the sub-list for extension extendee
|
|||
0, // [0:0] is the sub-list for field type_name
|
|||
} |
|||
|
|||
func init() { file_dtmgrpc_workflow_wfpb_wf_proto_init() } |
|||
func file_dtmgrpc_workflow_wfpb_wf_proto_init() { |
|||
if File_dtmgrpc_workflow_wfpb_wf_proto != nil { |
|||
return |
|||
} |
|||
if !protoimpl.UnsafeEnabled { |
|||
file_dtmgrpc_workflow_wfpb_wf_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} { |
|||
switch v := v.(*WorkflowData); i { |
|||
case 0: |
|||
return &v.state |
|||
case 1: |
|||
return &v.sizeCache |
|||
case 2: |
|||
return &v.unknownFields |
|||
default: |
|||
return nil |
|||
} |
|||
} |
|||
} |
|||
type x struct{} |
|||
out := protoimpl.TypeBuilder{ |
|||
File: protoimpl.DescBuilder{ |
|||
GoPackagePath: reflect.TypeOf(x{}).PkgPath(), |
|||
RawDescriptor: file_dtmgrpc_workflow_wfpb_wf_proto_rawDesc, |
|||
NumEnums: 0, |
|||
NumMessages: 1, |
|||
NumExtensions: 0, |
|||
NumServices: 1, |
|||
}, |
|||
GoTypes: file_dtmgrpc_workflow_wfpb_wf_proto_goTypes, |
|||
DependencyIndexes: file_dtmgrpc_workflow_wfpb_wf_proto_depIdxs, |
|||
MessageInfos: file_dtmgrpc_workflow_wfpb_wf_proto_msgTypes, |
|||
}.Build() |
|||
File_dtmgrpc_workflow_wfpb_wf_proto = out.File |
|||
file_dtmgrpc_workflow_wfpb_wf_proto_rawDesc = nil |
|||
file_dtmgrpc_workflow_wfpb_wf_proto_goTypes = nil |
|||
file_dtmgrpc_workflow_wfpb_wf_proto_depIdxs = nil |
|||
} |
|||
@ -0,0 +1,15 @@ |
|||
syntax = "proto3"; |
|||
|
|||
option go_package = "./wfpb"; |
|||
import "google/protobuf/empty.proto"; |
|||
|
|||
package workflow; |
|||
|
|||
// The Workflow service definition. |
|||
service Workflow { |
|||
rpc Execute(WorkflowData) returns (google.protobuf.Empty) {} |
|||
} |
|||
|
|||
message WorkflowData { |
|||
bytes Data = 1; |
|||
} |
|||
@ -0,0 +1,106 @@ |
|||
// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
|
|||
// versions:
|
|||
// - protoc-gen-go-grpc v1.2.0
|
|||
// - protoc v3.17.3
|
|||
// source: dtmgrpc/workflow/wfpb/wf.proto
|
|||
|
|||
package wfpb |
|||
|
|||
import ( |
|||
context "context" |
|||
grpc "google.golang.org/grpc" |
|||
codes "google.golang.org/grpc/codes" |
|||
status "google.golang.org/grpc/status" |
|||
emptypb "google.golang.org/protobuf/types/known/emptypb" |
|||
) |
|||
|
|||
// This is a compile-time assertion to ensure that this generated file
|
|||
// is compatible with the grpc package it is being compiled against.
|
|||
// Requires gRPC-Go v1.32.0 or later.
|
|||
const _ = grpc.SupportPackageIsVersion7 |
|||
|
|||
// WorkflowClient is the client API for Workflow service.
|
|||
//
|
|||
// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
|
|||
type WorkflowClient interface { |
|||
Execute(ctx context.Context, in *WorkflowData, opts ...grpc.CallOption) (*emptypb.Empty, error) |
|||
} |
|||
|
|||
type workflowClient struct { |
|||
cc grpc.ClientConnInterface |
|||
} |
|||
|
|||
func NewWorkflowClient(cc grpc.ClientConnInterface) WorkflowClient { |
|||
return &workflowClient{cc} |
|||
} |
|||
|
|||
func (c *workflowClient) Execute(ctx context.Context, in *WorkflowData, opts ...grpc.CallOption) (*emptypb.Empty, error) { |
|||
out := new(emptypb.Empty) |
|||
err := c.cc.Invoke(ctx, "/workflow.Workflow/Execute", in, out, opts...) |
|||
if err != nil { |
|||
return nil, err |
|||
} |
|||
return out, nil |
|||
} |
|||
|
|||
// WorkflowServer is the server API for Workflow service.
|
|||
// All implementations must embed UnimplementedWorkflowServer
|
|||
// for forward compatibility
|
|||
type WorkflowServer interface { |
|||
Execute(context.Context, *WorkflowData) (*emptypb.Empty, error) |
|||
mustEmbedUnimplementedWorkflowServer() |
|||
} |
|||
|
|||
// UnimplementedWorkflowServer must be embedded to have forward compatible implementations.
|
|||
type UnimplementedWorkflowServer struct { |
|||
} |
|||
|
|||
func (UnimplementedWorkflowServer) Execute(context.Context, *WorkflowData) (*emptypb.Empty, error) { |
|||
return nil, status.Errorf(codes.Unimplemented, "method Execute not implemented") |
|||
} |
|||
func (UnimplementedWorkflowServer) mustEmbedUnimplementedWorkflowServer() {} |
|||
|
|||
// UnsafeWorkflowServer may be embedded to opt out of forward compatibility for this service.
|
|||
// Use of this interface is not recommended, as added methods to WorkflowServer will
|
|||
// result in compilation errors.
|
|||
type UnsafeWorkflowServer interface { |
|||
mustEmbedUnimplementedWorkflowServer() |
|||
} |
|||
|
|||
func RegisterWorkflowServer(s grpc.ServiceRegistrar, srv WorkflowServer) { |
|||
s.RegisterService(&Workflow_ServiceDesc, srv) |
|||
} |
|||
|
|||
func _Workflow_Execute_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { |
|||
in := new(WorkflowData) |
|||
if err := dec(in); err != nil { |
|||
return nil, err |
|||
} |
|||
if interceptor == nil { |
|||
return srv.(WorkflowServer).Execute(ctx, in) |
|||
} |
|||
info := &grpc.UnaryServerInfo{ |
|||
Server: srv, |
|||
FullMethod: "/workflow.Workflow/Execute", |
|||
} |
|||
handler := func(ctx context.Context, req interface{}) (interface{}, error) { |
|||
return srv.(WorkflowServer).Execute(ctx, req.(*WorkflowData)) |
|||
} |
|||
return interceptor(ctx, in, info, handler) |
|||
} |
|||
|
|||
// Workflow_ServiceDesc is the grpc.ServiceDesc for Workflow service.
|
|||
// It's only intended for direct use with grpc.RegisterService,
|
|||
// and not to be introspected or modified (even as a copy)
|
|||
var Workflow_ServiceDesc = grpc.ServiceDesc{ |
|||
ServiceName: "workflow.Workflow", |
|||
HandlerType: (*WorkflowServer)(nil), |
|||
Methods: []grpc.MethodDesc{ |
|||
{ |
|||
MethodName: "Execute", |
|||
Handler: _Workflow_Execute_Handler, |
|||
}, |
|||
}, |
|||
Streams: []grpc.StreamDesc{}, |
|||
Metadata: "dtmgrpc/workflow/wfpb/wf.proto", |
|||
} |
|||
@ -0,0 +1,190 @@ |
|||
package workflow |
|||
|
|||
import ( |
|||
"context" |
|||
"database/sql" |
|||
"fmt" |
|||
"net/url" |
|||
|
|||
"github.com/dtm-labs/dtm/dtmcli" |
|||
"github.com/dtm-labs/dtm/dtmcli/dtmimp" |
|||
"github.com/dtm-labs/dtm/dtmcli/logger" |
|||
"github.com/dtm-labs/dtm/dtmgrpc/dtmgimp" |
|||
"github.com/dtm-labs/dtm/dtmgrpc/workflow/wfpb" |
|||
"github.com/go-resty/resty/v2" |
|||
"google.golang.org/grpc" |
|||
) |
|||
|
|||
// InitHttp will init Workflow engine to use http
|
|||
// param httpDtm specify the dtm address
|
|||
// param callback specify the url for dtm to callback if a workflow timeout
|
|||
func InitHttp(httpDtm string, callback string) { |
|||
defaultFac.protocol = dtmimp.ProtocolHTTP |
|||
defaultFac.httpDtm = httpDtm |
|||
defaultFac.httpCallback = callback |
|||
} |
|||
|
|||
// InitHttp will init Workflow engine to use grpc
|
|||
// param dtm specify the dtm address
|
|||
// param clientHost specify the client host for dtm to callback if a workflow timeout
|
|||
// param grpcServer specify the grpc server
|
|||
func InitGrpc(grpcDtm string, clientHost string, grpcServer *grpc.Server) { |
|||
defaultFac.protocol = dtmimp.ProtocolGRPC |
|||
defaultFac.grpcDtm = grpcDtm |
|||
wfpb.RegisterWorkflowServer(grpcServer, &workflowServer{}) |
|||
defaultFac.grpcCallback = clientHost + "/workflow.Workflow/Execute" |
|||
} |
|||
|
|||
// SetProtocolForTest change protocol directly. only used by test
|
|||
func SetProtocolForTest(protocol string) { |
|||
defaultFac.protocol = protocol |
|||
} |
|||
|
|||
// Register will register a workflow with the specified name
|
|||
func Register(name string, handler WfFunc) error { |
|||
return defaultFac.register(name, handler) |
|||
} |
|||
|
|||
// Execute will execute a workflow with the gid and specified params
|
|||
// if the workflow with the gid does not exist, then create a new workflow and execute it
|
|||
// if the workflow with the gid exists, resume to execute it
|
|||
func Execute(name string, gid string, data []byte) error { |
|||
return defaultFac.execute(name, gid, data) |
|||
} |
|||
|
|||
// ExecuteByQS is like Execute, but name and gid will be obtained from qs
|
|||
func ExecuteByQS(qs url.Values, body []byte) error { |
|||
return defaultFac.executeByQS(qs, body) |
|||
} |
|||
|
|||
// WorkflowOptions is for specifying workflow options
|
|||
type WorkflowOptions struct { |
|||
// if this flag is set true, then Workflow's restyClient will keep the origin http response
|
|||
// or else, Workflow's restyClient will convert http reponse to error if status code is not 200
|
|||
DisalbeAutoError bool |
|||
} |
|||
|
|||
// Workflow is the type for a workflow
|
|||
type Workflow struct { |
|||
// The name of the workflow
|
|||
Name string |
|||
Options WorkflowOptions |
|||
*dtmimp.TransBase |
|||
workflowImp |
|||
} |
|||
|
|||
// WfFunc is the type for workflow function
|
|||
type WfFunc func(wf *Workflow, data []byte) error |
|||
|
|||
// WfPhase2Func is the type for phase 2 function
|
|||
// param bb is a BranchBarrier, which is introduced by http://d.dtm.pub/practice/barrier.html
|
|||
type WfPhase2Func func(bb *dtmcli.BranchBarrier) error |
|||
|
|||
// NewRequest return a new resty request, whose progress will be recorded
|
|||
func (wf *Workflow) NewRequest() *resty.Request { |
|||
return wf.restyClient.R().SetContext(wf.Context) |
|||
} |
|||
|
|||
// DefineSagaPhase2 will define a saga branch transaction
|
|||
// param compensate specify a function for the compensation of next workflow action
|
|||
func (wf *Workflow) DefineSagaPhase2(compensate WfPhase2Func) { |
|||
branchID := wf.currentBranch |
|||
wf.failedOps = append(wf.failedOps, workflowPhase2Item{ |
|||
branchID: branchID, |
|||
op: dtmimp.OpRollback, |
|||
fn: compensate, |
|||
}) |
|||
} |
|||
|
|||
// DefineSagaPhase2 will define a tcc branch transaction
|
|||
// param confirm, concel specify the confirm and cancel operation of next workflow action
|
|||
func (wf *Workflow) DefineTccPhase2(confirm, cancel WfPhase2Func) { |
|||
branchID := wf.currentBranch |
|||
wf.failedOps = append(wf.failedOps, workflowPhase2Item{ |
|||
branchID: branchID, |
|||
op: dtmimp.OpRollback, |
|||
fn: cancel, |
|||
}) |
|||
wf.succeededOps = append(wf.succeededOps, workflowPhase2Item{ |
|||
branchID: branchID, |
|||
op: dtmimp.OpCommit, |
|||
fn: confirm, |
|||
}) |
|||
} |
|||
|
|||
// DoAction will do an action which will be recored
|
|||
func (wf *Workflow) DoAction(fn func(bb *dtmcli.BranchBarrier) ([]byte, error)) ([]byte, error) { |
|||
res := wf.recordedDo(func(bb *dtmcli.BranchBarrier) *stepResult { |
|||
r, e := fn(bb) |
|||
return stepResultFromLocal(r, e) |
|||
}) |
|||
return stepResultToLocal(res) |
|||
} |
|||
|
|||
func (wf *Workflow) DoXaAction(dbConf dtmcli.DBConf, fn func(db *sql.DB) ([]byte, error)) ([]byte, error) { |
|||
branchID := wf.currentBranch |
|||
res := wf.recordedDo(func(bb *dtmcli.BranchBarrier) *stepResult { |
|||
sBusi := "business" |
|||
k := bb.BranchID + "-" + sBusi |
|||
if wf.progresses[k] != nil { |
|||
return &stepResult{ |
|||
Error: fmt.Errorf("error occur at prepare, not resumable, to rollback. %w", dtmcli.ErrFailure), |
|||
} |
|||
} |
|||
sr := &stepResult{} |
|||
wf.TransBase.BranchID = branchID |
|||
wf.TransBase.Op = sBusi |
|||
err := dtmimp.XaHandleLocalTrans(wf.TransBase, dbConf, func(d *sql.DB) error { |
|||
r, e := fn(d) |
|||
sr.Data = r |
|||
if e == nil { |
|||
e = wf.saveResult(branchID, sBusi, &stepResult{Status: dtmcli.StatusSucceed}) |
|||
} |
|||
return e |
|||
}) |
|||
sr.Error = err |
|||
sr.Status = wfErrorToStatus(err) |
|||
return sr |
|||
}) |
|||
phase2 := func(bb *dtmcli.BranchBarrier) error { |
|||
return dtmimp.XaHandlePhase2(bb.Gid, dbConf, bb.BranchID, bb.Op) |
|||
} |
|||
wf.succeededOps = append(wf.succeededOps, workflowPhase2Item{ |
|||
branchID: branchID, |
|||
op: dtmimp.OpCommit, |
|||
fn: phase2, |
|||
}) |
|||
wf.failedOps = append(wf.failedOps, workflowPhase2Item{ |
|||
branchID: branchID, |
|||
op: dtmimp.OpRollback, |
|||
fn: phase2, |
|||
}) |
|||
return res.Data, res.Error |
|||
} |
|||
|
|||
// Interceptor is the middleware for workflow to capture grpc call result
|
|||
func Interceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error { |
|||
logger.Debugf("grpc client calling: %s%s %v", cc.Target(), method, dtmimp.MustMarshalString(req)) |
|||
wf := ctx.Value(wfMeta{}).(*Workflow) |
|||
|
|||
origin := func() error { |
|||
ctx1 := dtmgimp.TransInfo2Ctx(ctx, wf.Gid, wf.TransType, wf.currentBranch, wf.currentOp, wf.Dtm) |
|||
err := invoker(ctx1, method, req, reply, cc, opts...) |
|||
res := fmt.Sprintf("grpc client called: %s%s %s result: %s err: %v", |
|||
cc.Target(), method, dtmimp.MustMarshalString(req), dtmimp.MustMarshalString(reply), err) |
|||
if err != nil { |
|||
logger.Errorf("%s", res) |
|||
} else { |
|||
logger.Debugf("%s", res) |
|||
} |
|||
return err |
|||
} |
|||
if wf.currentOp != dtmimp.OpAction { |
|||
return origin() |
|||
} |
|||
sr := wf.recordedDo(func(bb *dtmcli.BranchBarrier) *stepResult { |
|||
err := origin() |
|||
return stepResultFromGrpc(reply, err) |
|||
}) |
|||
return stepResultToGrpc(sr, reply) |
|||
} |
|||
@ -0,0 +1,43 @@ |
|||
package dtmsvr |
|||
|
|||
import ( |
|||
"github.com/dtm-labs/dtm/dtmcli" |
|||
"github.com/dtm-labs/dtm/dtmcli/dtmimp" |
|||
"github.com/dtm-labs/dtm/dtmgrpc/dtmgimp" |
|||
"github.com/dtm-labs/dtm/dtmgrpc/workflow/wfpb" |
|||
) |
|||
|
|||
type transWorkflowProcessor struct { |
|||
*TransGlobal |
|||
} |
|||
|
|||
func init() { |
|||
registorProcessorCreator("workflow", func(trans *TransGlobal) transProcessor { return &transWorkflowProcessor{TransGlobal: trans} }) |
|||
} |
|||
|
|||
func (t *transWorkflowProcessor) GenBranches() []TransBranch { |
|||
return []TransBranch{} |
|||
} |
|||
|
|||
type cWorkflowCustom struct { |
|||
Name string `json:"name"` |
|||
Data []byte `json:"data"` |
|||
} |
|||
|
|||
func (t *transWorkflowProcessor) ProcessOnce(branches []TransBranch) error { |
|||
if t.Status == dtmcli.StatusSubmitted { // client workflow finished
|
|||
t.changeStatus(dtmcli.StatusSucceed) |
|||
return nil |
|||
} else if t.Status == dtmcli.StatusFailed || t.Status == dtmcli.StatusSucceed { |
|||
return nil |
|||
} |
|||
|
|||
cmc := cWorkflowCustom{} |
|||
dtmimp.MustUnmarshalString(t.CustomData, &cmc) |
|||
data := cmc.Data |
|||
if t.Protocol == dtmimp.ProtocolGRPC { |
|||
wd := wfpb.WorkflowData{Data: cmc.Data} |
|||
data = dtmgimp.MustProtoMarshal(&wd) |
|||
} |
|||
return t.getURLResult(t.QueryPrepared, "00", cmc.Name, data) |
|||
} |
|||
@ -1,13 +1,15 @@ |
|||
set -x |
|||
echo "" > coverage.txt |
|||
for store in redis mysql boltdb postgres; do |
|||
echo "mode: count" coverage.txt |
|||
for store in redis boltdb mysql postgres; do |
|||
for d in $(go list ./... | grep -v vendor); do |
|||
TEST_STORE=$store go test -covermode count -coverprofile=profile.out -coverpkg=github.com/dtm-labs/dtm/dtmcli,github.com/dtm-labs/dtm/dtmcli/dtmimp,github.com/dtm-labs/dtm/dtmcli/logger,github.com/dtm-labs/dtm/dtmgrpc,github.com/dtm-labs/dtm/dtmgrpc/dtmgimp,github.com/dtm-labs/dtm/dtmsvr,github.com/dtm-labs/dtm/dtmsvr/config,github.com/dtm-labs/dtm/dtmsvr/storage,github.com/dtm-labs/dtm/dtmsvr/storage/boltdb,github.com/dtm-labs/dtm/dtmsvr/storage/redis,github.com/dtm-labs/dtm/dtmsvr/storage/registry,github.com/dtm-labs/dtm/dtmsvr/storage/sql,github.com/dtm-labs/dtm/dtmutil -gcflags=-l $d || exit 1 |
|||
TEST_STORE=$store go test -covermode count -coverprofile=profile.out -coverpkg=github.com/dtm-labs/dtm/dtmcli,github.com/dtm-labs/dtm/dtmcli/dtmimp,github.com/dtm-labs/dtm/dtmcli/logger,github.com/dtm-labs/dtm/dtmgrpc,github.com/dtm-labs/dtm/dtmgrpc/workflow,github.com/dtm-labs/dtm/dtmgrpc/dtmgimp,github.com/dtm-labs/dtm/dtmsvr,github.com/dtm-labs/dtm/dtmsvr/config,github.com/dtm-labs/dtm/dtmsvr/storage,github.com/dtm-labs/dtm/dtmsvr/storage/boltdb,github.com/dtm-labs/dtm/dtmsvr/storage/redis,github.com/dtm-labs/dtm/dtmsvr/storage/registry,github.com/dtm-labs/dtm/dtmsvr/storage/sql,github.com/dtm-labs/dtm/dtmutil -gcflags=-l $d || exit 1 |
|||
if [ -f profile.out ]; then |
|||
cat profile.out >> coverage.txt |
|||
cat profile.out | grep -v 'mode:' >> coverage.txt |
|||
echo > profile.out |
|||
fi |
|||
done |
|||
done |
|||
|
|||
# go tool cover -html=coverage.txt |
|||
|
|||
curl -s https://codecov.io/bash | bash |
|||
|
|||
@ -0,0 +1,12 @@ |
|||
package busi |
|||
|
|||
import ( |
|||
"github.com/dtm-labs/dtm/dtmgrpc/workflow" |
|||
"github.com/dtm-labs/dtm/dtmutil" |
|||
"google.golang.org/grpc" |
|||
) |
|||
|
|||
func WorkflowStarup(server *grpc.Server) { |
|||
workflow.InitHttp(dtmServer, Busi+"/workflow/resume") |
|||
workflow.InitGrpc(dtmutil.DefaultGrpcServer, BusiGrpc, server) |
|||
} |
|||
@ -1,31 +1,14 @@ |
|||
package busi |
|||
|
|||
import ( |
|||
"context" |
|||
"fmt" |
|||
|
|||
"github.com/dtm-labs/dtm/dtmcli/dtmimp" |
|||
"github.com/dtm-labs/dtm/dtmutil" |
|||
"github.com/gin-gonic/gin" |
|||
) |
|||
|
|||
// Startup startup the busi's grpc and http service
|
|||
func Startup() *gin.Engine { |
|||
GrpcStartup() |
|||
return BaseAppStartup() |
|||
} |
|||
|
|||
// PopulateDB populate example mysql data
|
|||
func PopulateDB(skipDrop bool) { |
|||
resetXaData() |
|||
file := fmt.Sprintf("%s/busi.%s.sql", dtmutil.GetSQLDir(), BusiConf.Driver) |
|||
dtmutil.RunSQLScript(BusiConf, file, skipDrop) |
|||
file = fmt.Sprintf("%s/dtmcli.barrier.%s.sql", dtmutil.GetSQLDir(), BusiConf.Driver) |
|||
dtmutil.RunSQLScript(BusiConf, file, skipDrop) |
|||
file = fmt.Sprintf("%s/dtmsvr.storage.%s.sql", dtmutil.GetSQLDir(), BusiConf.Driver) |
|||
dtmutil.RunSQLScript(BusiConf, file, skipDrop) |
|||
_, err := RedisGet().FlushAll(context.Background()).Result() // redis barrier need clear
|
|||
dtmimp.E2P(err) |
|||
SetRedisBothAccount(10000, 10000) |
|||
SetupMongoBarrierAndBusi() |
|||
svr := GrpcStartup() |
|||
app := BaseAppStartup() |
|||
WorkflowStarup(svr) |
|||
go GrpcServe(svr) |
|||
return app |
|||
} |
|||
|
|||
@ -0,0 +1,265 @@ |
|||
/* |
|||
* Copyright (c) 2021 yedf. All rights reserved. |
|||
* Use of this source code is governed by a BSD-style |
|||
* license that can be found in the LICENSE file. |
|||
*/ |
|||
|
|||
package test |
|||
|
|||
import ( |
|||
"database/sql" |
|||
"testing" |
|||
|
|||
"github.com/dtm-labs/dtm/dtmcli" |
|||
"github.com/dtm-labs/dtm/dtmcli/dtmimp" |
|||
"github.com/dtm-labs/dtm/dtmcli/logger" |
|||
"github.com/dtm-labs/dtm/dtmgrpc/dtmgimp" |
|||
"github.com/dtm-labs/dtm/dtmgrpc/workflow" |
|||
"github.com/dtm-labs/dtm/test/busi" |
|||
"github.com/stretchr/testify/assert" |
|||
) |
|||
|
|||
func TestWorkflowNormal(t *testing.T) { |
|||
workflow.SetProtocolForTest(dtmimp.ProtocolHTTP) |
|||
req := busi.GenTransReq(30, false, false) |
|||
gid := dtmimp.GetFuncName() |
|||
|
|||
workflow.Register(gid, func(wf *workflow.Workflow, data []byte) error { |
|||
var req busi.TransReq |
|||
dtmimp.MustUnmarshal(data, &req) |
|||
_, err := wf.NewRequest().SetBody(req).Post(Busi + "/TransOut") |
|||
if err != nil { |
|||
return err |
|||
} |
|||
_, err = wf.NewRequest().SetBody(req).Post(Busi + "/TransIn") |
|||
if err != nil { |
|||
return err |
|||
} |
|||
return nil |
|||
}) |
|||
|
|||
err := workflow.Execute(gid, gid, dtmimp.MustMarshal(req)) |
|||
assert.Nil(t, err) |
|||
waitTransProcessed(gid) |
|||
assert.Equal(t, StatusSucceed, getTransStatus(gid)) |
|||
} |
|||
|
|||
func TestWorkflowRollback(t *testing.T) { |
|||
workflow.SetProtocolForTest(dtmimp.ProtocolHTTP) |
|||
|
|||
req := busi.GenTransReq(30, false, true) |
|||
gid := dtmimp.GetFuncName() |
|||
|
|||
workflow.Register(gid, func(wf *workflow.Workflow, data []byte) error { |
|||
var req busi.TransReq |
|||
dtmimp.MustUnmarshal(data, &req) |
|||
wf.DefineSagaPhase2(func(bb *dtmcli.BranchBarrier) error { |
|||
_, err := wf.NewRequest().SetBody(req).Post(Busi + "/SagaBTransOutCom") |
|||
return err |
|||
}) |
|||
_, err := wf.DoAction(func(bb *dtmcli.BranchBarrier) ([]byte, error) { |
|||
return nil, bb.CallWithDB(dbGet().ToSQLDB(), func(tx *sql.Tx) error { |
|||
return busi.SagaAdjustBalance(tx, busi.TransOutUID, -req.Amount, "") |
|||
}) |
|||
}) |
|||
if err != nil { |
|||
return err |
|||
} |
|||
wf.DefineSagaPhase2(func(bb *dtmcli.BranchBarrier) error { |
|||
return bb.CallWithDB(dbGet().ToSQLDB(), func(tx *sql.Tx) error { |
|||
return busi.SagaAdjustBalance(tx, busi.TransInUID, -req.Amount, "") |
|||
}) |
|||
}) |
|||
_, err = wf.NewRequest().SetBody(req).Post(Busi + "/SagaBTransIn") |
|||
if err != nil { |
|||
return err |
|||
} |
|||
return nil |
|||
}) |
|||
|
|||
err := workflow.Execute(gid, gid, dtmimp.MustMarshal(req)) |
|||
assert.Error(t, err, dtmcli.ErrFailure) |
|||
assert.Equal(t, StatusFailed, getTransStatus(gid)) |
|||
waitTransProcessed(gid) |
|||
} |
|||
|
|||
func TestWorkflowGrpcNormal(t *testing.T) { |
|||
workflow.SetProtocolForTest(dtmimp.ProtocolGRPC) |
|||
req := &busi.BusiReq{Amount: 30, TransInResult: "FAILURE"} |
|||
gid := dtmimp.GetFuncName() |
|||
workflow.Register(gid, func(wf *workflow.Workflow, data []byte) error { |
|||
var req busi.BusiReq |
|||
dtmgimp.MustProtoUnmarshal(data, &req) |
|||
wf.DefineSagaPhase2(func(bb *dtmcli.BranchBarrier) error { |
|||
_, err := busi.BusiCli.TransOutRevertBSaga(wf.Context, &req) |
|||
return err |
|||
}) |
|||
_, err := busi.BusiCli.TransOutBSaga(wf.Context, &req) |
|||
if err != nil { |
|||
return err |
|||
} |
|||
wf.DefineSagaPhase2(func(bb *dtmcli.BranchBarrier) error { |
|||
_, err := busi.BusiCli.TransInRevertBSaga(wf.Context, &req) |
|||
return err |
|||
}) |
|||
_, err = busi.BusiCli.TransInBSaga(wf.Context, &req) |
|||
return err |
|||
}) |
|||
err := workflow.Execute(gid, gid, dtmgimp.MustProtoMarshal(req)) |
|||
assert.Error(t, err, dtmcli.ErrFailure) |
|||
assert.Equal(t, StatusFailed, getTransStatus(gid)) |
|||
waitTransProcessed(gid) |
|||
} |
|||
|
|||
var ongoingStep = 0 |
|||
|
|||
func fetchOngoingStep(dest int) bool { |
|||
c := ongoingStep |
|||
logger.Debugf("ongoing step is: %d", c) |
|||
if c == dest { |
|||
ongoingStep++ |
|||
return true |
|||
} |
|||
return false |
|||
} |
|||
|
|||
func TestWorkflowGrpcRollbackResume(t *testing.T) { |
|||
workflow.SetProtocolForTest(dtmimp.ProtocolGRPC) |
|||
gid := dtmimp.GetFuncName() |
|||
ongoingStep = 0 |
|||
workflow.Register(gid, func(wf *workflow.Workflow, data []byte) error { |
|||
var req busi.BusiReq |
|||
dtmgimp.MustProtoUnmarshal(data, &req) |
|||
if fetchOngoingStep(0) { |
|||
return dtmcli.ErrOngoing |
|||
} |
|||
wf.DefineSagaPhase2(func(bb *dtmcli.BranchBarrier) error { |
|||
if fetchOngoingStep(4) { |
|||
return dtmcli.ErrOngoing |
|||
} |
|||
_, err := busi.BusiCli.TransOutRevertBSaga(wf.Context, &req) |
|||
return err |
|||
}) |
|||
_, err := busi.BusiCli.TransOutBSaga(wf.Context, &req) |
|||
if fetchOngoingStep(1) { |
|||
return dtmcli.ErrOngoing |
|||
} |
|||
if err != nil { |
|||
return err |
|||
} |
|||
wf.DefineSagaPhase2(func(bb *dtmcli.BranchBarrier) error { |
|||
if fetchOngoingStep(3) { |
|||
return dtmcli.ErrOngoing |
|||
} |
|||
_, err := busi.BusiCli.TransInRevertBSaga(wf.Context, &req) |
|||
return err |
|||
}) |
|||
_, err = busi.BusiCli.TransInBSaga(wf.Context, &req) |
|||
if fetchOngoingStep(2) { |
|||
return dtmcli.ErrOngoing |
|||
} |
|||
return err |
|||
}) |
|||
req := &busi.BusiReq{Amount: 30, TransInResult: "FAILURE"} |
|||
err := workflow.Execute(gid, gid, dtmgimp.MustProtoMarshal(req)) |
|||
assert.Error(t, err, dtmcli.ErrOngoing) |
|||
assert.Equal(t, StatusPrepared, getTransStatus(gid)) |
|||
cronTransOnceForwardNow(t, gid, 1000) |
|||
assert.Equal(t, StatusPrepared, getTransStatus(gid)) |
|||
cronTransOnceForwardNow(t, gid, 1000) |
|||
assert.Equal(t, StatusPrepared, getTransStatus(gid)) |
|||
cronTransOnceForwardNow(t, gid, 1000) |
|||
assert.Equal(t, StatusPrepared, getTransStatus(gid)) |
|||
cronTransOnceForwardNow(t, gid, 1000) |
|||
assert.Equal(t, StatusPrepared, getTransStatus(gid)) |
|||
// next cron will make a workflow submit, and do an additional write to chan, so make an additional read chan
|
|||
go waitTransProcessed(gid) |
|||
cronTransOnceForwardNow(t, gid, 1000) |
|||
assert.Equal(t, StatusFailed, getTransStatus(gid)) |
|||
} |
|||
|
|||
func TestWorkflowXaAction(t *testing.T) { |
|||
workflow.SetProtocolForTest(dtmimp.ProtocolGRPC) |
|||
gid := dtmimp.GetFuncName() |
|||
workflow.Register(gid, func(wf *workflow.Workflow, data []byte) error { |
|||
_, err := wf.DoXaAction(busi.BusiConf, func(db *sql.DB) ([]byte, error) { |
|||
return nil, busi.SagaAdjustBalance(db, busi.TransOutUID, -30, dtmcli.ResultSuccess) |
|||
}) |
|||
if err != nil { |
|||
return err |
|||
} |
|||
_, err = wf.DoXaAction(busi.BusiConf, func(db *sql.DB) ([]byte, error) { |
|||
return nil, busi.SagaAdjustBalance(db, busi.TransInUID, 30, dtmcli.ResultSuccess) |
|||
}) |
|||
return err |
|||
}) |
|||
err := workflow.Execute(gid, gid, nil) |
|||
assert.Nil(t, err) |
|||
waitTransProcessed(gid) |
|||
assert.Equal(t, StatusSucceed, getTransStatus(gid)) |
|||
} |
|||
|
|||
func TestWorkflowXaRollback(t *testing.T) { |
|||
workflow.SetProtocolForTest(dtmimp.ProtocolGRPC) |
|||
gid := dtmimp.GetFuncName() |
|||
workflow.Register(gid, func(wf *workflow.Workflow, data []byte) error { |
|||
_, err := wf.DoXaAction(busi.BusiConf, func(db *sql.DB) ([]byte, error) { |
|||
return nil, busi.SagaAdjustBalance(db, busi.TransOutUID, -30, dtmcli.ResultSuccess) |
|||
}) |
|||
if err != nil { |
|||
return err |
|||
} |
|||
_, err = wf.DoXaAction(busi.BusiConf, func(db *sql.DB) ([]byte, error) { |
|||
e := busi.SagaAdjustBalance(db, busi.TransInUID, 30, dtmcli.ResultSuccess) |
|||
logger.FatalIfError(e) |
|||
return nil, dtmcli.ErrFailure |
|||
}) |
|||
return err |
|||
}) |
|||
err := workflow.Execute(gid, gid, nil) |
|||
assert.Equal(t, dtmcli.ErrFailure, err) |
|||
waitTransProcessed(gid) |
|||
assert.Equal(t, StatusFailed, getTransStatus(gid)) |
|||
} |
|||
|
|||
func TestWorkflowXaResume(t *testing.T) { |
|||
workflow.SetProtocolForTest(dtmimp.ProtocolGRPC) |
|||
ongoingStep = 0 |
|||
gid := dtmimp.GetFuncName() |
|||
workflow.Register(gid, func(wf *workflow.Workflow, data []byte) error { |
|||
_, err := wf.DoXaAction(busi.BusiConf, func(db *sql.DB) ([]byte, error) { |
|||
if fetchOngoingStep(0) { |
|||
return nil, dtmcli.ErrOngoing |
|||
} |
|||
return nil, busi.SagaAdjustBalance(db, busi.TransOutUID, -30, dtmcli.ResultSuccess) |
|||
}) |
|||
if err != nil { |
|||
return err |
|||
} |
|||
_, err = wf.DoXaAction(busi.BusiConf, func(db *sql.DB) ([]byte, error) { |
|||
if fetchOngoingStep(1) { |
|||
return nil, dtmcli.ErrOngoing |
|||
} |
|||
return nil, busi.SagaAdjustBalance(db, busi.TransInUID, 30, dtmcli.ResultSuccess) |
|||
}) |
|||
if err != nil { |
|||
return err |
|||
} |
|||
if fetchOngoingStep(2) { |
|||
return dtmcli.ErrOngoing |
|||
} |
|||
|
|||
return err |
|||
}) |
|||
err := workflow.Execute(gid, gid, nil) |
|||
assert.Equal(t, dtmcli.ErrOngoing, err) |
|||
|
|||
cronTransOnceForwardNow(t, gid, 1000) |
|||
assert.Equal(t, StatusPrepared, getTransStatus(gid)) |
|||
cronTransOnceForwardNow(t, gid, 1000) |
|||
assert.Equal(t, StatusPrepared, getTransStatus(gid)) |
|||
// next cron will make a workflow submit, and do an additional write to chan, so make an additional read chan
|
|||
go waitTransProcessed(gid) |
|||
cronTransOnceForwardNow(t, gid, 1000) |
|||
assert.Equal(t, StatusSucceed, getTransStatus(gid)) |
|||
} |
|||
Loading…
Reference in new issue