support join annotation

script/uploadToServer.sh

@@ -0,0 +1,3 @@
+#/bin/bash
+
+scp -P 222 ./build/bdserver.zip dev@47.95.110.68:/data/public/releases/bdcontract/$1/

src/main/java/org/bdware/server/executor/unconsistency/MultiPointCooperationExecutor.java

@@ -8,10 +8,7 @@ import org.apache.logging.log4j.Logger;
 import org.bdware.sc.ComponedContractResult;
 import org.bdware.sc.ContractMeta;
 import org.bdware.sc.ContractResult;
-import org.bdware.sc.bean.ContractExecType;
+import org.bdware.sc.bean.*;
-import org.bdware.sc.bean.ContractRequest;
-import org.bdware.sc.bean.FunctionDesp;
-import org.bdware.sc.bean.RouteInfo;
 import org.bdware.sc.conn.OnHashCallback;
 import org.bdware.sc.conn.ResultCallback;
 import org.bdware.sc.units.MultiContractMeta;
@@ -64,12 +61,12 @@ public class MultiPointCooperationExecutor implements ContractExecutor {
             final ResultCallback originalCb,
             final int count,
             final int request_seq,
-            final String contractID) {
+            final String contractID, JoinInfo joinInfo) {
-        ComponedContractResult componedContractResult = new ComponedContractResult(count);
         // TODO 加对应的超时？
+
         return new ResultCollector(
                 requestID,
-                new MultiPointCooperationExecutor.ResultMerger(originalCb, count, request_seq, contractID),
+                new MultiPointCooperationExecutor.ResultMerger(originalCb, count, request_seq, contractID, joinInfo),
                 count); // 把count改成了1，设置成获得1个响应就行
     }
 
@@ -89,39 +86,19 @@ public class MultiPointCooperationExecutor implements ContractExecutor {
         MasterServerTCPAction.requestCache.get(contractID).put(req.seq, sendStr);
 
         LOGGER.debug(JsonUtil.toJson(req));
-
         String[] nodes = getAccordingToRouteInfo(req);
+
         LOGGER.info("node size = " + nodes.length);
         LOGGER.debug("nodes:" + JsonUtil.toJson(nodes));
         for (String node : nodes) {
-            LOGGER.info(node);
+            LOGGER.info(
-            LOGGER.info("[sendRequests] get cmNode " + node.substring(0, 5));
+                    "[sendRequests] get cmNode "
-            if (!NetworkManager.instance.hasAgentConnection(node)) {
+                            + node.substring(0, 5)
-                LOGGER.info("[sendRequests] get cmNode " + node.substring(0, 5) + " is null");
+                            + " not null "
-                collector.onResult(
+                            + "RequestAllExecutor  发送请求给 "
-                        "{\"status\":\"Error\",\"result\":\"node offline\","
+                            + node.substring(0, 5));
-                                + "\"nodeID\":\""
+            NetworkManager.instance.sendToAgent(node, sendStr);
-                                + node
+
-                                + "\","
-                                + "\"action\":\"onExecuteContractTrustfully\"}");
-            } else if (MasterServerRecoverMechAction.recoverStatus.get(node).get(contractID)
-                    != RecoverFlag.Fine) {
-                collector.onResult(
-                        "{\"status\":\"Error\",\"result\":\"node recovering\","
-                                + "\"nodeID\":\""
-                                + node
-                                + "\","
-                                + "\"action\":\"onExecuteContractTrustfully\"}");
-                NetworkManager.instance.sendToAgent(node, sendStr);
-            } else {
-                LOGGER.info(
-                        "[sendRequests] get cmNode "
-                                + node.substring(0, 5)
-                                + " not null "
-                                + "RequestAllExecutor  发送请求给 "
-                                + node.substring(0, 5));
-                NetworkManager.instance.sendToAgent(node, sendStr);
-            }
         }
     }
 
@@ -224,15 +201,16 @@ public class MultiPointCooperationExecutor implements ContractExecutor {
                     CMActions.manager.statusRecorder.getContractMeta(req.getContractID());
             FunctionDesp fun = meta.getExportedFunction(req.getAction());
             ResultCallback collector;
-            // TODO @huyueyang，这个ResultCallback可能要根据@Join来。
+            // TODO @fanbo 下面的count 1要改，应该是根据route的规则来。
+            JoinInfo joinInfo = fun.joinInfo;
             if (fun.getRoute() != null && fun.getRoute().useDefault != null) {
                 // 根据函数的类型决定不同的 collector
                 collector =
-                        createResultCallback(id, rc, 1, req.seq, req.getContractID()); // 初始化结果收集器
+                        createResultCallback(id, rc, 1, req.seq, req.getContractID(), joinInfo); // 初始化结果收集器
             } else {
                 collector =
                         createResultCallback(
-                                id, rc, resultCount, req.seq, req.getContractID()); // 初始化结果收集器
+                                id, rc, resultCount, req.seq, req.getContractID(), joinInfo); // 初始化结果收集器
             }
 
             MasterServerTCPAction.sync.sleep(id, collector);
@@ -271,18 +249,21 @@ public class MultiPointCooperationExecutor implements ContractExecutor {
         int request_seq;
         ResultCallback originalCallback;
         Set<String> nodeIDs = new HashSet<>(); // 已收到返回结果的节点
+        JoinInfo joinInfo;
 
         ResultMerger(
                 final ResultCallback originalCb,
                 final int count,
                 final int request_seq,
-                final String contractID) {
+                final String contractID,
+                final JoinInfo joinInfo) {
             originalCallback = originalCb;
             this.count = count;
             this.request_seq = request_seq;
             this.contractID = contractID;
             componedContractResult = new ComponedContractResult(count);
             order = new AtomicInteger(0);
+            this.joinInfo = joinInfo;
         }
 
         public String getInfo() {
@@ -327,6 +308,7 @@ public class MultiPointCooperationExecutor implements ContractExecutor {
                 // 收集到所有结果
                 if (order.incrementAndGet() == count) {
                     ContractResult finalResult = componedContractResult.mergeFinalResult();
+
                     finalResult.needSeq = true;
                     finalResult.seq = request_seq;
 
@@ -340,6 +322,9 @@ public class MultiPointCooperationExecutor implements ContractExecutor {
                     //                        originalCallback.onResult(new
                     // Gson().toJson(finalResult));
                     //                    } else {
+                    if (joinInfo != null) {
+                        handleJoinInfo(finalResult, joinInfo);
+                    }
                     originalCallback.onResult(JsonUtil.toJson(finalResult));
                     //                    }
                     LOGGER.info(
@@ -383,5 +368,29 @@ public class MultiPointCooperationExecutor implements ContractExecutor {
                 LOGGER.info("本次执行最终结果为有异常");
             }
         }
+
+        private void handleJoinInfo(ContractResult finalResult, JoinInfo joinInfo) {
+            JsonObject jo = finalResult.result.getAsJsonObject();
+            if (joinInfo != null && joinInfo.joinRule != null) {
+                //TODO 不应该是double 类型
+
+                switch (joinInfo.joinRule) {
+                    case add:
+                        double val = 0;
+                        for (String key : jo.keySet()) {
+                            val += jo.get(key).getAsDouble();
+                        }
+                        finalResult.result = new JsonPrimitive(val);
+                        break;
+                    case multiply:
+                        val = 1;
+                        for (String key : jo.keySet()) {
+                            val *= jo.get(key).getAsDouble();
+                        }
+                        finalResult.result = new JsonPrimitive(val);
+                        break;
+                }
+            }
+        }
     }
 }