concurrent timeout supported

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",

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)
-}

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.

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))
 	}

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)

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

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

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)

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

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

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,