Merge pull request #27 from catcherwong/saga

feat: support saga

Author: li-xiao-shuang

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>pub.dtm</groupId>
     <artifactId>dtmcli-java</artifactId>
-    <version>1.5.4-SNAPSHOT</version>
+    <version>1.5.5-SNAPSHOT</version>
 
     <properties>
         <maven.compiler.source>8</maven.compiler.source>
@@ -17,6 +17,7 @@
         <commons-lang3.version>3.12.0</commons-lang3.version>
         <slf4j-api.version>1.7.30</slf4j-api.version>
         <mysql-connector-java.version>8.0.27</mysql-connector-java.version>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     </properties>
 
     <dependencies>

src/main/java/client/DtmClient.java

@@ -30,6 +30,7 @@
 import common.model.DtmServerInfo;
 import common.utils.HttpUtil;
 import okhttp3.Response;
+import saga.Saga;
 import tcc.Tcc;
 
 import java.util.Objects;
@@ -38,13 +39,13 @@
  * @author lixiaoshuang
  */
 public class DtmClient {
-    
+
     private String ipPort;
-    
+
     public DtmClient(String ipPort) {
         this.ipPort = ipPort;
     }
-    
+
     /**
      * 生成全局事务id
      *
@@ -73,7 +74,7 @@ public String genGid() throws Exception {
         }
         return jsonObject.get("gid").toString();
     }
-    
+
     /**
      * tcc事务
      *
@@ -86,6 +87,13 @@ public void tccGlobalTransaction(String gid, DtmConsumer<Tcc> function) throws E
         Tcc tcc = new Tcc(ipPort, gid);
         tcc.tccGlobalTransaction(function);
     }
-    
-    
+
+    /**
+     * create a new saga
+     * @param gid
+     */
+    public Saga newSaga(String gid) {
+        Saga saga = new Saga(ipPort, gid);
+        return saga;
+    }
 }

src/main/java/common/constant/ParamFieldConstant.java

@@ -28,29 +28,49 @@
  * @author lixiaoshuang
  */
 public class ParamFieldConstant {
-    
+
     public static final String GID = "gid";
-    
+
     public static final String TRANS_TYPE = "trans_type";
-    
+
     public static final String BRANCH_ID = "branch_id";
-    
+
     public static final String STATUS = "status";
-    
+
     public static final String DATA = "data";
-    
+
     public static final String TRY = "try";
-    
+
     public static final String CONFIRM = "confirm";
-    
+
     public static final String CANCEL = "cancel";
-    
+
     public static final String OP = "op";
-    
+
     public static final String CODE = "code";
-    
+
     public static final String MESSAGE = "message";
-    
+
     public static final String DTM_RESULT = "dtm_result";
-    
+
+    public static final String ACTION = "action";
+
+    public static final String COMPENSATE = "compensate";
+
+    public static final String STEPS = "steps";
+
+    public static final String PAYLOADS = "payloads";
+
+    public static final String CUSTOM_DATA = "custom_data";
+
+    public static final String WAIT_RESULT = "wait_result";
+
+    public static final String TIMEOUT_TO_FAIL = "timeout_to_fail";
+
+    public static final String RETRY_INTERVAL = "retry_interval";
+
+    public static final String PASSTHROGH_HEADERS = "passthrough_headers";
+
+    public static final String BRANCH_HEADERS = "branch_headers";
+
 }

src/main/java/common/enums/TransTypeEnum.java

@@ -26,16 +26,16 @@
 
 public enum TransTypeEnum {
     // 事务类型
-    TCC("tcc");
-    
+    TCC("tcc"), SAGA("saga");
+
     TransTypeEnum(String value) {
         this.value = value;
     }
-    
+
     private String value;
-    
+
     public String getValue() {
         return this.value;
     }
-    
+
 }

src/main/java/common/model/TransBase.java

@@ -28,23 +28,40 @@
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Map;
+
 @Data
 @NoArgsConstructor
 public class TransBase {
-    
+
     /**
      * 全局事务id
      */
     private String gid;
-    
+
     /**
      * 事务类型
      */
     private TransTypeEnum transTypeEnum;
-    
-    
+
     private boolean waitResult;
-    
+
+    private long timeoutToFail;
+
+    private long retryInterval;
+
+    private Map<String, String> branchHeaders = new HashMap<>();
+
+    private ArrayList<String> passthroughHeaders = new ArrayList<>();
+
+    private String customData;
+
+    private ArrayList<Map<String, String>> steps = new ArrayList<>();
+
+    private ArrayList<String> payloads = new ArrayList<>();
+
     public TransBase(TransTypeEnum transTypeEnum, String gid, boolean waitResult) {
         this.gid = gid;
         this.transTypeEnum = transTypeEnum;

src/main/java/saga/Saga.java

@@ -24,6 +24,106 @@
 
 package saga;
 
+import com.alibaba.fastjson.JSONObject;
+import common.constant.Constant;
+import common.constant.ParamFieldConstant;
+import common.model.DtmServerInfo;
+import common.utils.HttpUtil;
+import common.enums.TransTypeEnum;
+import common.model.TransBase;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Map;
+
 public class Saga {
 
+    private static final String ORDERS = "orders";
+
+    private static final String CONCURRENT = "concurrent";
+
+    private TransBase transBase;
+
+    private DtmServerInfo dtmServerInfo;
+
+    private boolean concurrent;
+
+    private Map<String, ArrayList<Integer>> orders;
+
+    public Saga(String ipPort, String gid) {
+        this.dtmServerInfo = new DtmServerInfo(ipPort);
+        this.transBase = new TransBase(TransTypeEnum.SAGA, gid, false);
+        this.concurrent = false;
+        this.orders = new HashMap<>();
+    }
+
+    public Saga add(String action, String compensate, Object postData) {
+        HashMap<String, String> step = new HashMap<>();
+        step.put(ParamFieldConstant.ACTION, action);
+        step.put(ParamFieldConstant.COMPENSATE, compensate);
+        transBase.getSteps().add(step);
+        transBase.getPayloads().add(JSONObject.toJSONString(postData));
+        return this;
+    }
+
+    public Saga addBranchOrder(Integer branch, ArrayList<Integer> preBranches) {
+        orders.put(branch.toString(), preBranches);
+        return this;
+    }
+
+    public Saga enableConcurrent() {
+        concurrent = true;
+        return this;
+    }
+
+    public void submit() throws Exception {
+        addConcurrentContext();
+        HashMap<String, Object> mapParam = new HashMap<>(Constant.DEFAULT_INITIAL_CAPACITY);
+        mapParam.put(ParamFieldConstant.GID, transBase.getGid());
+        mapParam.put(ParamFieldConstant.TRANS_TYPE, TransTypeEnum.SAGA.getValue());
+        mapParam.put(ParamFieldConstant.STEPS, transBase.getSteps());
+        mapParam.put(ParamFieldConstant.PAYLOADS, transBase.getPayloads());
+        mapParam.put(ParamFieldConstant.CUSTOM_DATA, transBase.getCustomData());
+        mapParam.put(ParamFieldConstant.WAIT_RESULT, transBase.isWaitResult());
+        mapParam.put(ParamFieldConstant.TIMEOUT_TO_FAIL, transBase.getTimeoutToFail());
+        mapParam.put(ParamFieldConstant.RETRY_INTERVAL, transBase.getRetryInterval());
+        mapParam.put(ParamFieldConstant.PASSTHROGH_HEADERS, transBase.getPassthroughHeaders());
+        mapParam.put(ParamFieldConstant.BRANCH_HEADERS, transBase.getBranchHeaders());
+
+        HttpUtil.post(dtmServerInfo.submit(), JSONObject.toJSONString(mapParam));
+    }
+
+    public Saga enableWaitResult() {
+        transBase.setWaitResult(true);
+        return this;
+    }
+
+    public Saga setTimeoutToFail(long timeoutToFail) {
+        transBase.setTimeoutToFail(timeoutToFail);
+        return this;
+    }
+
+    public Saga setRetryInterval(long retryInterval) {
+        transBase.setRetryInterval(retryInterval);
+        return this;
+    }
+
+    public Saga setBranchHeaders(Map<String, String> headers) {
+        transBase.setBranchHeaders(headers);
+        return this;
+    }
+
+    public Saga setPassthroughHeaders(ArrayList<String> passthroughHeaders) {
+        transBase.setPassthroughHeaders(passthroughHeaders);
+        return this;
+    }
+
+    private void addConcurrentContext() {
+        if (concurrent) {
+            HashMap<String, Object> data = new HashMap<>();
+            data.put(ORDERS, orders);
+            data.put(CONCURRENT, true);
+            transBase.setCustomData(JSONObject.toJSONString(data));
+        }
+    }
 }