|
@@ -6,10 +6,9 @@ import org.bfkj.utils.LogUtils;
|
|
|
import org.bfkj.utils.MapTools;
|
|
|
import org.bfkj.utils.MyDbHelper;
|
|
|
import org.bfkj.utils.ScriptEnginePro;
|
|
|
-import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
|
|
|
|
|
|
import java.util.*;
|
|
|
-import java.util.concurrent.ThreadPoolExecutor;
|
|
|
+import java.util.stream.Collectors;
|
|
|
|
|
|
public class DataProcess {
|
|
|
private String serviceId;
|
|
@@ -28,7 +27,7 @@ public class DataProcess {
|
|
|
public DataProcess(String service_Id) {
|
|
|
lastActive = System.currentTimeMillis();
|
|
|
serviceId = service_Id;
|
|
|
- try{
|
|
|
+ try {
|
|
|
baseDbHelper = new MyDbHelper(AppConfig.getSystemParams(AppConfig.REMOTE_DB_CONNECT));
|
|
|
if (Objects.nonNull(baseDbHelper.getErrorMessage())) {
|
|
|
errorMessage = "获取底座myDbHelper对象异常: ".concat(baseDbHelper.getErrorMessage());
|
|
@@ -36,7 +35,7 @@ public class DataProcess {
|
|
|
}
|
|
|
|
|
|
Map<String, Object> calculationMap = baseDbHelper.queryByParamsReturnList("SELECT CL.*,DI.* FROM calculation_library CL left JOIN datasourceinfo DI ON DI.dataSourceID = CL.datasource_id WHERE CL.service_id =? and CL.library_type is not null order by library_sort,library_id", service_Id);
|
|
|
- calcList = Objects.isNull(calculationMap.get("returnData"))?null:(List<Map<String, Object>>) calculationMap.get("returnData");
|
|
|
+ calcList = Objects.isNull(calculationMap.get("returnData")) ? null : (List<Map<String, Object>>) calculationMap.get("returnData");
|
|
|
if (!calculationMap.get("code").equals("0") || calcList == null || calcList.isEmpty()) {
|
|
|
errorMessage = "查询".concat(serviceId).concat("的算法失败:").concat(calculationMap.get("message").toString());
|
|
|
return;
|
|
@@ -51,10 +50,11 @@ public class DataProcess {
|
|
|
if (!Objects.isNull(serviceAuthList)) {
|
|
|
serviceAuthMap = ((List<Map<String, Object>>) serviceAuthList).stream().map(map -> map.get("app_id").toString()).toList();
|
|
|
}
|
|
|
- }catch (Exception e) {
|
|
|
+ } catch (Exception e) {
|
|
|
errorMessage = "数据处理对象初始化异常: ".concat(serviceId).concat(";异常信息:").concat(LogUtils.getException(e));
|
|
|
}
|
|
|
}
|
|
|
+
|
|
|
|
|
|
public void close() {
|
|
|
for (ScriptEnginePro scriptEnginePro : ScriptEngineProMaps.values()) {
|
|
@@ -67,6 +67,7 @@ public class DataProcess {
|
|
|
calcDbHelperMaps.clear();
|
|
|
baseDbHelper.close();
|
|
|
}
|
|
|
+
|
|
|
|
|
|
public Map<String, Object> processSuccess(Object returnData) {
|
|
|
if (returnData instanceof Map) {
|
|
@@ -79,6 +80,7 @@ public class DataProcess {
|
|
|
returnMap.put("returnData", returnData);
|
|
|
return returnMap;
|
|
|
}
|
|
|
+
|
|
|
|
|
|
public Map<String, Object> processFail(String errorMessage, String libraryId) {
|
|
|
Map<String, Object> returnMap = new HashMap<>();
|
|
@@ -89,6 +91,7 @@ public class DataProcess {
|
|
|
}
|
|
|
return returnMap;
|
|
|
}
|
|
|
+
|
|
|
|
|
|
public Map<String, Object> processData(Map<String, Object> inputData, String... user_id) {
|
|
|
if (!MapTools.isBlank(errorMessage)) {
|
|
@@ -105,6 +108,9 @@ public class DataProcess {
|
|
|
}
|
|
|
String dataObjectId = inputData.get("dataObjectId").toString();
|
|
|
inputData = authCheck(inputData, user_id);
|
|
|
+ if ("-1".equals(inputData.get("code")) || (Objects.nonNull( inputData.get("message")) && inputData.get("message").toString().equals("服务未授任何数据权限"))) {
|
|
|
+ return inputData;
|
|
|
+ }
|
|
|
List<Map<String, Object>> calcData = new ArrayList<>();
|
|
|
calcData.add(inputData);
|
|
|
Map<String, Object> execResult = execCalultion(calcData, null, dataObjectId);
|
|
@@ -112,16 +118,23 @@ public class DataProcess {
|
|
|
List<Map<String, Object>> preListData = (List<Map<String, Object>>) execResult.get("preData");
|
|
|
calcData = (List<Map<String, Object>>) execResult.get("calcData");
|
|
|
|
|
|
- Map<String, Object> lastResult = library_id.endsWith("N")?preListData.get(preListData.size() - 1):calcData.get(calcData.size() - 1);
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ Map<String, Object> lastResult = library_id.endsWith("N") ? preListData.get(preListData.size() - 1) : calcData.get(calcData.size() - 1);
|
|
|
if (Objects.equals(lastResult.get("code"), "-1")) {
|
|
|
Object message = lastResult.get("message");
|
|
|
- LogUtils.log("processData:3", "-1", library_id, Objects.nonNull(message)?message.toString():null, serviceId, calcData, dataObjectId, preListData, inputData.get("event"));
|
|
|
+ LogUtils.log("processData:3", "-1", library_id, Objects.nonNull(message) ? message.toString() : null, serviceId, calcData, dataObjectId, preListData, inputData.get("event"));
|
|
|
return processFail(library_id.concat("数据计算错误:").concat(lastResult.get("message").toString()), library_id);
|
|
|
}
|
|
|
|
|
|
LogUtils.log("DataProcess:9999", "0", library_id, "数据接收后处理成功", serviceId, calcData, dataObjectId, preListData, inputData.get("event"));
|
|
|
- return processSuccess(lastResult);
|
|
|
+
|
|
|
+ calcData = calcData.stream().filter(Objects::nonNull).collect(Collectors.toList());
|
|
|
+ return processSuccess(calcData.get(calcData.size() - 1));
|
|
|
}
|
|
|
+
|
|
|
|
|
|
public Map<String, Object> authCheck(Map<String, Object> inputData, String... user_id) {
|
|
|
Object appid = inputData.get("appid");
|
|
@@ -160,7 +173,7 @@ public class DataProcess {
|
|
|
rowAuth.get(rowAuth.size() - 1).put("connector", " and ");
|
|
|
rowAuth.get(rowAuth.size() - 1).put("right", " ) ");
|
|
|
}
|
|
|
- }
|
|
|
+ }
|
|
|
});
|
|
|
if (rowAuth.size() > 0) {
|
|
|
rowAuth.get(0).put("left", "(".concat(rowAuth.get(0).get("left").toString()));
|
|
@@ -172,6 +185,7 @@ public class DataProcess {
|
|
|
}
|
|
|
return inputData;
|
|
|
}
|
|
|
+
|
|
|
|
|
|
public Map<String, Object> execCalultion(List<Map<String, Object>> inData, String beginLibraryId, String dataObjectId) {
|
|
|
Map<String, Object> returnData = new HashMap<>();
|
|
@@ -216,6 +230,7 @@ public class DataProcess {
|
|
|
returnData.put("calcData", calcData);
|
|
|
return returnData;
|
|
|
}
|
|
|
+
|
|
|
|
|
|
private Map<String, Object> execEngine(String library_id, Map<String, Object> currentCalMap, List<Map<String, Object>> calcAllData) {
|
|
|
if (!ScriptEngineProMaps.containsKey(library_id)) {
|
|
@@ -239,15 +254,16 @@ public class DataProcess {
|
|
|
|
|
|
|
|
|
|
|
|
- if (currentCalMap.get("library_type").toString().equals("2")){
|
|
|
+ if (currentCalMap.get("library_type").toString().equals("2")) {
|
|
|
paramValue.put(parmaNames.get(key), currentData);
|
|
|
- }else {
|
|
|
+ } else {
|
|
|
paramValue.put(key, currentData);
|
|
|
}
|
|
|
|
|
|
}
|
|
|
return currentEngin.execScript(paramValue);
|
|
|
}
|
|
|
+
|
|
|
|
|
|
private Map<String, Object> execDB(String library_id, Map<String, Object> calculationLibrary, List<Map<String, Object>> calcData, String dataObjectId) {
|
|
|
Object connectConfigObj = calculationLibrary.get("connectConfig");
|
|
@@ -269,35 +285,36 @@ public class DataProcess {
|
|
|
return processFail("获取业务数据库对象失败".concat(errMessage), library_id);
|
|
|
}
|
|
|
|
|
|
- String paramIndex = Objects.isNull(calculationLibrary.get("parmIndex")) ? (String.valueOf(calcData.size() - 1)).concat(".returnData") : calculationLibrary.get("parmIndex").toString();
|
|
|
+ String paramIndex = Objects.isNull(calculationLibrary.get("parmIndex"))?(String.valueOf(calcData.size() - 1)).concat(calcData.size() ==1?".dataContent":".returnData") : calculationLibrary.get("parmIndex").toString();
|
|
|
Object tempDBPrams = dataSubscription(calcData, "List.".concat(paramIndex), calculationLibrary);
|
|
|
|
|
|
|
|
|
- if("7".equals(calculationLibrary.get("event")) && !Objects.isNull(calculationLibrary.get("tableName")) && !Objects.isNull(calculationLibrary.get("computing_expression"))){
|
|
|
- Map<String,Object> tempCalcInfo = calculationLibrary;
|
|
|
+ if ("7".equals(calculationLibrary.get("event")) && !Objects.isNull(calculationLibrary.get("tableName")) && !Objects.isNull(calculationLibrary.get("computing_expression"))) {
|
|
|
+ Map<String, Object> tempCalcInfo = calculationLibrary;
|
|
|
String tableName = tempCalcInfo.get("tableName").toString();
|
|
|
String sqlStr = tempCalcInfo.get("computing_expression").toString();
|
|
|
- tempCalcInfo.put("tableName","");
|
|
|
+ tempCalcInfo.put("tableName", "");
|
|
|
tempDBPrams = myDbHelper.generalProcess(tempCalcInfo, tempDBPrams, dataObjectId, calcData.get(0));
|
|
|
|
|
|
|
|
|
- tempCalcInfo.put("tableName",tableName);
|
|
|
- tempCalcInfo.put("computing_expression","");
|
|
|
- tempCalcInfo.put("event","6");
|
|
|
+ tempCalcInfo.put("tableName", tableName);
|
|
|
+ tempCalcInfo.put("computing_expression", "");
|
|
|
+ tempCalcInfo.put("event", "6");
|
|
|
return myDbHelper.generalProcess(tempCalcInfo, tempDBPrams, dataObjectId, calcData.get(0));
|
|
|
}
|
|
|
return myDbHelper.generalProcess(calculationLibrary, tempDBPrams, dataObjectId, calcData.get(0));
|
|
|
}
|
|
|
+
|
|
|
|
|
|
private Object dataSubscription(List<Map<String, Object>> calcAllData, String paramRule, Map<String, Object> calculationLibrary) {
|
|
|
String[] itemRule = paramRule.split("\\.");
|
|
|
String dataType = itemRule.length > 0 ? itemRule[0] : "List";
|
|
|
- dataType = dataType.endsWith("]")?dataType.substring(0,dataType.indexOf("[")):dataType;
|
|
|
+ dataType = dataType.endsWith("]") ? dataType.substring(0, dataType.indexOf("[")) : dataType;
|
|
|
String dataLocation = itemRule.length > 1 ? itemRule[1] : "";
|
|
|
if (dataLocation.equals("T")) {
|
|
|
- Object tempObj = itemRule.length > 2 ? MapTools.isBlank(itemRule[2])?("List".equals(dataType)?new ArrayList<>():("JSON,Map".contains(dataType)?new HashMap<>():null)):itemRule[2]:null;
|
|
|
- if ("Boolean".equals(dataType)){
|
|
|
- tempObj = itemRule[2].equals("true");
|
|
|
+ Object tempObj = itemRule.length > 2 ? MapTools.isBlank(itemRule[2]) ? ("List".equals(dataType) ? new ArrayList<>() : ("JSON,Map".contains(dataType) ? new HashMap<>() : null)) : itemRule[2] : null;
|
|
|
+ if ("Boolean".equals(dataType)) {
|
|
|
+ tempObj = itemRule[2].equals("true");
|
|
|
}
|
|
|
return tempObj;
|
|
|
|
|
@@ -305,7 +322,7 @@ public class DataProcess {
|
|
|
|
|
|
Object returnData = dataLocation.equals("L") ? calculationLibrary : MapTools.isNumber(dataLocation) ? calcAllData.get(Integer.parseInt(dataLocation)) : null;
|
|
|
for (int index = 2; index < itemRule.length; index++) {
|
|
|
- if(Objects.isNull(returnData)) return null;
|
|
|
+ if (Objects.isNull(returnData)) return null;
|
|
|
if (returnData instanceof String) {
|
|
|
returnData = MapTools.strToObj(returnData.toString());
|
|
|
}
|
|
@@ -313,10 +330,10 @@ public class DataProcess {
|
|
|
|
|
|
returnData = returnData instanceof List ? ((List<?>) returnData).get(Integer.parseInt(itemRule[index])) : returnData;
|
|
|
} else {
|
|
|
- if(returnData instanceof Map){
|
|
|
- returnData = ((Map<?, ?>) returnData).get(itemRule[index]);
|
|
|
- }else{
|
|
|
- if(returnData instanceof List && ((List<?>) returnData).size() > 0 && ((List<?>) returnData).get(0) instanceof Map){
|
|
|
+ if (returnData instanceof Map) {
|
|
|
+ returnData = itemRule[index].equals("returnData") && Objects.isNull(((Map<?, ?>) returnData).get("returnData")) ? ((Map<?, ?>) returnData).get("dataContent") : ((Map<?, ?>) returnData).get(itemRule[index]);
|
|
|
+ } else {
|
|
|
+ if (returnData instanceof List && ((List<?>) returnData).size() > 0 && ((List<?>) returnData).get(0) instanceof Map) {
|
|
|
returnData = ((Map<?, ?>) ((List<?>) returnData).get(0)).get(itemRule[index]);
|
|
|
}
|
|
|
}
|
|
@@ -342,11 +359,12 @@ public class DataProcess {
|
|
|
}
|
|
|
return returnData;
|
|
|
}
|
|
|
+
|
|
|
|
|
|
public void setServiceErrorCount(String library_id, String success) {
|
|
|
if (!success.equals("0")) {
|
|
|
- Integer errCount = Objects.isNull(serviceErrorCount.get(library_id))?1:serviceErrorCount.get(library_id)+1;
|
|
|
- serviceErrorCount.put(library_id,errCount);
|
|
|
+ Integer errCount = Objects.isNull(serviceErrorCount.get(library_id)) ? 1 : serviceErrorCount.get(library_id) + 1;
|
|
|
+ serviceErrorCount.put(library_id, errCount);
|
|
|
if (errCount > AppConfig.SERVICE_ERR_MAX) {
|
|
|
ServiceInputControl.stop(serviceId);
|
|
|
}
|
|
@@ -354,8 +372,10 @@ public class DataProcess {
|
|
|
serviceErrorCount.put(library_id, 0);
|
|
|
}
|
|
|
}
|
|
|
+
|
|
|
private long sequence = 0L;
|
|
|
private long lastTimestamp = -1L;
|
|
|
+
|
|
|
public String createLifeCycleCol(Long workerId, Integer serviceId) {
|
|
|
long timestamp = System.currentTimeMillis();
|
|
|
|