diff --git a/client/dtmcli/dtmimp/trans_base.go b/client/dtmcli/dtmimp/trans_base.go index 16ecbd8..67e4acb 100644 --- a/client/dtmcli/dtmimp/trans_base.go +++ b/client/dtmcli/dtmimp/trans_base.go @@ -99,10 +99,8 @@ func TransCallDtmExt(tb *TransBase, body interface{}, operation string) (*resty. if tb.Protocol == Jrpc { return transCallDtmJrpc(tb, body, operation) } - if tb.RequestTimeout != 0 { - RestyClient.SetTimeout(time.Duration(tb.RequestTimeout) * time.Second) - } - resp, err := RestyClient.R(). + rc := GetRestyClient2(time.Duration(tb.RequestTimeout) * time.Second) + resp, err := rc.R(). SetBody(body).Post(fmt.Sprintf("%s/%s", tb.Dtm, operation)) if err != nil { return nil, err @@ -147,7 +145,7 @@ func TransRequestBranch(t *TransBase, method string, body interface{}, branchID if t.TransType == "xa" { // xa trans will add notify_url query["phase2_url"] = url } - resp, err := RestyClient.R(). + resp, err := GetRestyClient2(0).R(). SetBody(body). SetQueryParams(query). SetHeaders(t.BranchHeaders). @@ -156,11 +154,9 @@ func TransRequestBranch(t *TransBase, method string, body interface{}, branchID } func transCallDtmJrpc(tb *TransBase, body interface{}, operation string) (*resty.Response, error) { - if tb.RequestTimeout != 0 { - RestyClient.SetTimeout(time.Duration(tb.RequestTimeout) * time.Second) - } + rc := GetRestyClient2(time.Duration(tb.RequestTimeout) * time.Second) var result map[string]interface{} - resp, err := RestyClient.R(). + resp, err := rc.R(). SetBody(map[string]interface{}{ "jsonrpc": "2.0", "id": "no-use", diff --git a/client/dtmcli/dtmimp/vars.go b/client/dtmcli/dtmimp/vars.go index 14259dd..7ac6918 100644 --- a/client/dtmcli/dtmimp/vars.go +++ b/client/dtmcli/dtmimp/vars.go @@ -8,6 +8,8 @@ package dtmimp import ( "errors" + "sync" + "time" "github.com/dtm-labs/dtmdriver" "github.com/dtm-labs/logger" @@ -30,15 +32,25 @@ var MapSuccess = map[string]interface{}{"dtm_result": ResultSuccess} // MapFailure HTTP result of FAILURE var MapFailure = map[string]interface{}{"dtm_result": ResultFailure} -// RestyClient the resty object -var RestyClient = resty.New() - // PassthroughHeaders will be passed to every sub-trans call var PassthroughHeaders = []string{} // BarrierTableName the table name of barrier table var BarrierTableName = "dtm_barrier.barrier" +var restyClients sync.Map + +func GetRestyClient2(timeout time.Duration) *resty.Client { + cli, ok := restyClients.Load(timeout) + if !ok { + client := resty.New() + AddRestyMiddlewares(client) + restyClients.Store(timeout, client) + cli = client + } + return cli.(*resty.Client) +} + // AddRestyMiddlewares will add the middlewares used by dtm func AddRestyMiddlewares(client *resty.Client) { client.OnBeforeRequest(func(c *resty.Client, r *resty.Request) error { @@ -58,7 +70,3 @@ func AddRestyMiddlewares(client *resty.Client) { return nil }) } - -func init() { - AddRestyMiddlewares(RestyClient) -} diff --git a/client/dtmcli/types.go b/client/dtmcli/types.go index be7fda1..48aa77b 100644 --- a/client/dtmcli/types.go +++ b/client/dtmcli/types.go @@ -7,6 +7,8 @@ package dtmcli import ( + "time" + "github.com/dtm-labs/dtm/client/dtmcli/dtmimp" "github.com/go-resty/resty/v2" ) @@ -37,7 +39,11 @@ func SetBarrierTableName(tablename string) { // GetRestyClient get the resty.Client for http request func GetRestyClient() *resty.Client { - return dtmimp.RestyClient + return dtmimp.GetRestyClient2(0) +} + +func GetRestyClient2(timeout time.Duration) *resty.Client { + return dtmimp.GetRestyClient2(timeout) } // SetPassthroughHeaders experimental. diff --git a/client/dtmcli/utils.go b/client/dtmcli/utils.go index 983edd7..0a3ef5c 100644 --- a/client/dtmcli/utils.go +++ b/client/dtmcli/utils.go @@ -13,7 +13,7 @@ import ( // MustGenGid generate a new gid func MustGenGid(server string) string { res := map[string]string{} - resp, err := dtmimp.RestyClient.R().SetResult(&res).Get(server + "/newGid") + resp, err := GetRestyClient().R().SetResult(&res).Get(server + "/newGid") if err != nil || res["gid"] == "" { panic(fmt.Errorf("newGid error: %v, resp: %s", err, resp)) } diff --git a/client/workflow/rpc.go b/client/workflow/rpc.go index 29dc96e..a6c63e0 100644 --- a/client/workflow/rpc.go +++ b/client/workflow/rpc.go @@ -3,6 +3,7 @@ package workflow import ( "context" + "github.com/dtm-labs/dtm/client/dtmcli" "github.com/dtm-labs/dtm/client/dtmcli/dtmimp" "github.com/dtm-labs/dtm/client/dtmgrpc/dtmgimp" "github.com/dtm-labs/dtm/client/dtmgrpc/dtmgpb" @@ -19,7 +20,7 @@ func (wf *Workflow) getProgress() ([]*dtmgpb.DtmProgress, error) { } return nil, err } - resp, err := dtmimp.RestyClient.R().SetBody(wf.TransBase).Post(wf.Dtm + "/prepareWorkflow") + resp, err := dtmcli.GetRestyClient().R().SetBody(wf.TransBase).Post(wf.Dtm + "/prepareWorkflow") var reply dtmgpb.DtmProgressesReply if err == nil { dtmimp.MustUnmarshal(resp.Body(), &reply) diff --git a/dtmsvr/trans_status.go b/dtmsvr/trans_status.go index 9453a8c..fb89e7f 100644 --- a/dtmsvr/trans_status.go +++ b/dtmsvr/trans_status.go @@ -109,9 +109,6 @@ func (t *TransGlobal) getURLResult(uri string, branchID, op string, branchPayloa return nil } if t.Protocol == dtmimp.ProtocolHTTP || strings.HasPrefix(uri, "http://") || strings.HasPrefix(uri, "https://") { - if t.RequestTimeout != 0 { - dtmimp.RestyClient.SetTimeout(time.Duration(t.RequestTimeout) * time.Second) - } if t.Protocol == "json-rpc" && strings.Contains(uri, "method") { return t.getJSONRPCResult(uri, branchID, op, branchPayload) } @@ -121,7 +118,8 @@ func (t *TransGlobal) getURLResult(uri string, branchID, op string, branchPayloa } func (t *TransGlobal) getHTTPResult(uri string, branchID, op string, branchPayload []byte) error { - resp, err := dtmimp.RestyClient.R().SetBody(string(branchPayload)). + rc := dtmimp.GetRestyClient2(time.Duration(t.RequestTimeout) * time.Second) + resp, err := rc.R().SetBody(string(branchPayload)). SetQueryParams(map[string]string{ "gid": t.Gid, "trans_type": t.TransType, @@ -147,7 +145,8 @@ func (t *TransGlobal) getJSONRPCResult(uri string, branchID, op string, branchPa params["trans_type"] = t.TransType params["branch_id"] = branchID params["op"] = op - resp, err := dtmimp.RestyClient.R().SetBody(map[string]interface{}{ + rc := dtmimp.GetRestyClient2(time.Duration(t.RequestTimeout) * time.Second) + resp, err := rc.R().SetBody(map[string]interface{}{ "params": params, "jsonrpc": "2.0", "method": u.Query().Get("method"), diff --git a/helper/bench/svr/http.go b/helper/bench/svr/http.go index 0761eeb..45a3b34 100644 --- a/helper/bench/svr/http.go +++ b/helper/bench/svr/http.go @@ -175,9 +175,9 @@ func benchAddRoute(app *gin.Engine) { err := saga.Submit() dtmimp.E2P(err) } else { - _, err := dtmimp.RestyClient.R().SetBody(gin.H{}).SetQueryParam("uid", suid2).Post(benchBusi + "/TransOut") + _, err := dtmcli.GetRestyClient().R().SetBody(gin.H{}).SetQueryParam("uid", suid2).Post(benchBusi + "/TransOut") dtmimp.E2P(err) - _, err = dtmimp.RestyClient.R().SetBody(gin.H{}).SetQueryParam("uid", suid).Post(benchBusi + "/TransIn") + _, err = dtmcli.GetRestyClient().R().SetBody(gin.H{}).SetQueryParam("uid", suid).Post(benchBusi + "/TransIn") dtmimp.E2P(err) } return nil diff --git a/test/api_test.go b/test/api_test.go index 6b58584..3c0b9ae 100644 --- a/test/api_test.go +++ b/test/api_test.go @@ -12,6 +12,7 @@ import ( "strconv" "testing" + "github.com/dtm-labs/dtm/client/dtmcli" "github.com/dtm-labs/dtm/client/dtmcli/dtmimp" "github.com/dtm-labs/dtm/dtmutil" "github.com/dtm-labs/dtm/test/busi" @@ -19,7 +20,7 @@ import ( ) func TestAPIVersion(t *testing.T) { - resp, err := dtmimp.RestyClient.R().Get(dtmutil.DefaultHTTPServer + "/version") + resp, err := dtmcli.GetRestyClient().R().Get(dtmutil.DefaultHTTPServer + "/version") assert.Nil(t, err) assert.Equal(t, 200, resp.StatusCode()) } @@ -29,7 +30,7 @@ func TestAPIQuery(t *testing.T) { err := genMsg(gid).Submit() assert.Nil(t, err) waitTransProcessed(gid) - resp, err := dtmimp.RestyClient.R().SetQueryParam("gid", gid).Get(dtmutil.DefaultHTTPServer + "/query") + resp, err := dtmcli.GetRestyClient().R().SetQueryParam("gid", gid).Get(dtmutil.DefaultHTTPServer + "/query") assert.Nil(t, err) m := map[string]interface{}{} assert.Equal(t, resp.StatusCode(), 200) @@ -37,11 +38,11 @@ func TestAPIQuery(t *testing.T) { assert.NotEqual(t, nil, m["transaction"]) assert.Equal(t, 2, len(m["branches"].([]interface{}))) - resp, err = dtmimp.RestyClient.R().SetQueryParam("gid", "").Get(dtmutil.DefaultHTTPServer + "/query") + resp, err = dtmcli.GetRestyClient().R().SetQueryParam("gid", "").Get(dtmutil.DefaultHTTPServer + "/query") e2p(err) assert.Equal(t, resp.StatusCode(), 500) - resp, err = dtmimp.RestyClient.R().SetQueryParam("gid", "1").Get(dtmutil.DefaultHTTPServer + "/query") + resp, err = dtmcli.GetRestyClient().R().SetQueryParam("gid", "1").Get(dtmutil.DefaultHTTPServer + "/query") e2p(err) assert.Equal(t, resp.StatusCode(), 200) dtmimp.MustUnmarshalString(resp.String(), &m) @@ -56,14 +57,14 @@ func TestAPIAll(t *testing.T) { assert.Nil(t, err) waitTransProcessed(gid) } - resp, err := dtmimp.RestyClient.R().SetQueryParam("limit", "1").Get(dtmutil.DefaultHTTPServer + "/all") + resp, err := dtmcli.GetRestyClient().R().SetQueryParam("limit", "1").Get(dtmutil.DefaultHTTPServer + "/all") assert.Nil(t, err) m := map[string]interface{}{} dtmimp.MustUnmarshalString(resp.String(), &m) nextPos := m["next_position"].(string) assert.NotEqual(t, "", nextPos) - resp, err = dtmimp.RestyClient.R().SetQueryParams(map[string]string{ + resp, err = dtmcli.GetRestyClient().R().SetQueryParams(map[string]string{ "limit": "1", "position": nextPos, }).Get(dtmutil.DefaultHTTPServer + "/all") @@ -73,7 +74,7 @@ func TestAPIAll(t *testing.T) { assert.NotEqual(t, "", nextPos2) assert.NotEqual(t, nextPos, nextPos2) - resp, err = dtmimp.RestyClient.R().SetQueryParams(map[string]string{ + resp, err = dtmcli.GetRestyClient().R().SetQueryParams(map[string]string{ "limit": "1000", "position": nextPos, }).Get(dtmutil.DefaultHTTPServer + "/all") @@ -84,7 +85,7 @@ func TestAPIAll(t *testing.T) { } func TestDtmMetrics(t *testing.T) { - rest, err := dtmimp.RestyClient.R().Get("http://localhost:36789/api/metrics") + rest, err := dtmcli.GetRestyClient().R().Get("http://localhost:36789/api/metrics") assert.Nil(t, err) assert.Equal(t, rest.StatusCode(), 200) } @@ -94,7 +95,7 @@ func TestAPIResetCronTime(t *testing.T) { sTimeout := strconv.FormatInt(timeout, 10) sLimit := strconv.FormatInt(limit, 10) - resp, err := dtmimp.RestyClient.R().SetQueryParams(map[string]string{ + resp, err := dtmcli.GetRestyClient().R().SetQueryParams(map[string]string{ "timeout": sTimeout, "limit": sLimit, }).Get(dtmutil.DefaultHTTPServer + "/resetCronTime") @@ -116,7 +117,7 @@ func TestAPIForceStoppedNormal(t *testing.T) { waitTransProcessed(saga.Gid) assert.Equal(t, StatusSubmitted, getTransStatus(saga.Gid)) - resp, err := dtmimp.RestyClient.R().SetBody(map[string]string{ + resp, err := dtmcli.GetRestyClient().R().SetBody(map[string]string{ "gid": saga.Gid, }).Post(dtmutil.DefaultHTTPServer + "/forceStop") assert.Nil(t, err) @@ -131,7 +132,7 @@ func TestAPIForceStoppedAbnormal(t *testing.T) { assert.Equal(t, []string{StatusPrepared, StatusSucceed, StatusPrepared, StatusSucceed}, getBranchesStatus(saga.Gid)) assert.Equal(t, StatusSucceed, getTransStatus(saga.Gid)) - resp, err := dtmimp.RestyClient.R().SetBody(map[string]string{ + resp, err := dtmcli.GetRestyClient().R().SetBody(map[string]string{ "gid": saga.Gid, }).Post(dtmutil.DefaultHTTPServer + "/forceStop") assert.Nil(t, err) diff --git a/test/base_test.go b/test/base_test.go index 6dc727d..d660e18 100644 --- a/test/base_test.go +++ b/test/base_test.go @@ -61,10 +61,10 @@ func TestBaseSqlDB(t *testing.T) { } func TestBaseHttp(t *testing.T) { - resp, err := dtmimp.RestyClient.R().SetQueryParam("panic_string", "1").Post(busi.Busi + "/TestPanic") + resp, err := dtmcli.GetRestyClient().R().SetQueryParam("panic_string", "1").Post(busi.Busi + "/TestPanic") assert.Nil(t, err) assert.Contains(t, resp.String(), "panic_string") - resp, err = dtmimp.RestyClient.R().SetQueryParam("panic_error", "1").Post(busi.Busi + "/TestPanic") + resp, err = dtmcli.GetRestyClient().R().SetQueryParam("panic_error", "1").Post(busi.Busi + "/TestPanic") assert.Nil(t, err) assert.Contains(t, resp.String(), "panic_error") } diff --git a/test/saga_compatible_test.go b/test/saga_compatible_test.go index ce97319..b482c29 100644 --- a/test/saga_compatible_test.go +++ b/test/saga_compatible_test.go @@ -10,6 +10,7 @@ import ( "fmt" "testing" + "github.com/dtm-labs/dtm/client/dtmcli" "github.com/dtm-labs/dtm/client/dtmcli/dtmimp" "github.com/dtm-labs/dtm/dtmutil" "github.com/dtm-labs/dtm/test/busi" @@ -20,7 +21,7 @@ func TestSagaCompatibleNormal(t *testing.T) { // compatible with old http, which gid := dtmimp.GetFuncName() body := fmt.Sprintf(`{"gid":"%s","trans_type":"saga","steps":[{"action":"%s/TransOut","compensate":"%s/TransOutRevert","data":"{\"amount\":30,\"transInResult\":\"SUCCESS\",\"transOutResult\":\"SUCCESS\"}"},{"action":"%s/TransIn","compensate":"%s/TransInRevert","data":"{\"amount\":30,\"transInResult\":\"SUCCESS\",\"transOutResult\":\"SUCCESS\"}"}]}`, gid, busi.Busi, busi.Busi, busi.Busi, busi.Busi) - dtmimp.RestyClient.R().SetBody(body).Post(fmt.Sprintf("%s/submit", dtmutil.DefaultHTTPServer)) + dtmcli.GetRestyClient().R().SetBody(body).Post(fmt.Sprintf("%s/submit", dtmutil.DefaultHTTPServer)) waitTransProcessed(gid) assert.Equal(t, []string{StatusPrepared, StatusSucceed, StatusPrepared, StatusSucceed}, getBranchesStatus(gid)) assert.Equal(t, StatusSucceed, getTransStatus(gid)) diff --git a/test/tcc_barrier_test.go b/test/tcc_barrier_test.go index 171b429..b298dde 100644 --- a/test/tcc_barrier_test.go +++ b/test/tcc_barrier_test.go @@ -85,7 +85,7 @@ func runTestTccBarrierDisorder(t *testing.T, store string) { return res }) // register tcc branch - resp, err := dtmimp.RestyClient.R(). + resp, err := dtmcli.GetRestyClient().R(). SetBody(map[string]interface{}{ "gid": tcc.Gid, "branch_id": branchID, @@ -113,7 +113,7 @@ func runTestTccBarrierDisorder(t *testing.T, store string) { cancelCanReturnChan <- "1" logger.Debugf("after cancelCanRetrun 2 write") // after cancel then run try - r, _ := dtmimp.RestyClient.R(). + r, _ := dtmcli.GetRestyClient().R(). SetBody(body). SetQueryParams(map[string]string{ "dtm": tcc.Dtm,