bdware/cp
2
0
Fork 0
mirror of https://gitee.com/BDWare/cp.git synced 2025-01-09 17:34:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor doip setup process

Browse Source
This commit is contained in:
CaiHQ 2023-03-27 16:11:49 +08:00
parent cd87b1f014
commit 16f2d4fae1
5 changed files with 103 additions and 190 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
build.gradle
View File

@ -6,7 +6,7 @@ plugins {
}
group = "org.bdware.sc"
version = "1.8.0"
version = "1.8.1"
tasks.withType(JavaCompile) {
// options.compilerArgs << '-Xlint:none'
// options.compilerArgs << '-Xlint:deprecation' << "-Werror"
@ -48,7 +48,8 @@ dependencies {
implementation 'com.sun.mail:javax.mail:1.6.2'
implementation 'com.squareup.okhttp3:okhttp:4.9.1'
implementation 'org.bdware.bdcontract:sdk-java:1.0.2'
implementation 'org.bdware.doip:doip-audit-tool:1.1.3'
implementation 'org.bdware.doip:doip-audit-tool:1.1.4'
implementation 'org.bdware.doip:doip-sdk:1.3.8'
implementation fileTree(dir: 'lib', include: '*.jar')
testImplementation 'junit:junit:4.13.2'
}
@ -66,8 +67,8 @@ jar {
// uncomment this when publish,
// while develop at local use "false"
configurations.runtimeClasspath.filter {
it.getAbsolutePath().contains("/lib/")
// false
// it.getAbsolutePath().contains("/lib/")
false
}.collect {
it.isDirectory() ? it : zipTree(it)
}
@ -123,8 +124,7 @@ tasks.withType(Javadoc) {
task javadocJar(type: Jar, dependsOn: javadoc) {
archiveClassifier = 'javadoc'
classifier = "javadoc"
exclude {
details -> details.file.getAbsolutePath().contains("/gm/")
exclude { details -> details.file.getAbsolutePath().contains("/gm/")
}
from javadoc.destinationDir
}

101
src/main/java/org/bdware/sc/ContractProcess.java
View File

@ -50,7 +50,6 @@ import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.zip.ZipEntry;
import java.util.zip.ZipFile;
public class ContractProcess {
@ -506,10 +505,8 @@ public class ContractProcess {
}
handleLog();
// System.out.println("[ret.getManifest().getInsnLimit()]" +
// ret.getManifest().getInsnLimit());
engine = new DesktopEngine(zipBundle.getManifest(), zipPath, contract);
engine.loadJar(zf);
engine.registerResource(new Resources(zf, engine.getClassLoad()));
@ -520,14 +517,12 @@ public class ContractProcess {
jo.add("loadContract", JsonUtil.parseObject(result));
jo.addProperty("status", result.status.merge(onCreate.status).toString());
LOGGER.debug("result: " + jo.toString());
// doipModule的话，拉起DoipServer服务端口
if(cn.getYjsType() == YjsType.DoipModule) {
if (cn.hasDoipModule()) {
// 只有一台机器去更新Router中的repoInfo就可以了
updateRepoInfo(contract.getCreateParam());
invokeOnStartingDoipServer(cn, contract.getCreateParam());
invokeOnStartingDoipServer(cn, contract.getCreateParam(), jo);
}
return jo.toString();
} else {
contract.setScript(FileUtil.getFileContent(zipPath));
@ -564,14 +559,14 @@ public class ContractProcess {
if (fun.isConfidential()) {
fun.appendBeforeInvokeHandler(new ConfidentialHandler(fun));
}
ArgSchemaHandler argSchemaHandler = createHandlerIfExist(fun, fun.annotations, ArgSchemaHandler.class);
ArgSchemaHandler argSchemaHandler = createHandlerIfExist(fun, ArgSchemaHandler.class);
if (argSchemaHandler != null) {
fun.appendBeforeInvokeHandler(argSchemaHandler);
}
if (fun.isExport()) {
//if(fun.annotations...)
AccessHandler accessHandler = createHandlerIfExist(fun,fun.annotations,AccessHandler.class);
AccessHandler accessHandler = createHandlerIfExist(fun, AccessHandler.class);
if (accessHandler != null) {
fun.appendBeforeInvokeHandler(accessHandler);
}
@ -602,20 +597,17 @@ public class ContractProcess {
}
}
<T extends AnnotationHook> T createHandlerIfExist(FunctionNode function, List<AnnotationNode> annotations, Class<T> clz) {
<T extends AnnotationHook> T createHandlerIfExist(FunctionNode function, Class<T> clz) {
YJSAnnotation annotation = clz.getAnnotation(YJSAnnotation.class);
if (annotation == null) return null;
if (annotations == null) return null;
for (AnnotationNode node : annotations) {
if (annotation.name().equals(node.getType())) {
try {
Method m = clz.getDeclaredMethod("fromAnnotationNode", FunctionNode.class, AnnotationNode.class);
T result = (T) m.invoke(null, function, node);
return result;
} catch (Exception e) {
e.printStackTrace();
}
}
try {
AnnotationNode node = function.getAnnotation(annotation.name());
if (node == null) return null;
Method m = clz.getDeclaredMethod("fromAnnotationNode", FunctionNode.class, AnnotationNode.class);
T result = (T) m.invoke(null, function, node);
return result;
} catch (Exception e) {
e.printStackTrace();
}
return null;
}
@ -663,9 +655,9 @@ public class ContractProcess {
LOGGER.debug("result: " + jo.toString());
// doipModule的话，拉起DoipServer服务端口
if(cn.getYjsType() == YjsType.DoipModule) {
if (cn.getYjsType() == YjsType.DoipModule) {
updateRepoInfo(contract.getCreateParam());
invokeOnStartingDoipServer(cn, contract.getCreateParam());
invokeOnStartingDoipServer(cn, contract.getCreateParam(), jo);
}
return jo.toString();
@ -679,14 +671,15 @@ public class ContractProcess {
public void updateRepoInfo(JsonElement arg) throws Exception {
// 只有0号节点需要初始化IRP连接去updateRepoInfo
if(JavaScriptEntry.shardingID == 0) {
if (JavaScriptEntry.shardingID == 0) {
// DOOP relevant logic
DoipClusterServer server = DoipClusterServer.getDOOPServerInstance();
if(server == null) {
if (server == null) {
JsonObject createParams = arg.getAsJsonObject();
if(createParams.has("router")) {
if (createParams.has("router")) {
JsonElement routerInfo = createParams.get("router");
if(!routerInfo.isJsonObject()) throw new Exception("Provide wrong router info in create params to DoipModule");
if (!routerInfo.isJsonObject())
throw new Exception("Provide wrong router info in create params to DoipModule");
else {
EndpointConfig endpointConfig = JsonUtil.GSON.fromJson(routerInfo.getAsJsonObject(), EndpointConfig.class);
DoipClusterServer.createDOOPServerInstance(endpointConfig);
@ -697,13 +690,13 @@ public class ContractProcess {
server = DoipClusterServer.getDOOPServerInstance();
}
// 只有一台机器去更新Router中的repoInfo就可以了
server.updateRepoInfo(contract, cn);
}
}
public void invokeOnStartingDoipServer(ContractNode cn, JsonElement arg) {
public void invokeOnStartingDoipServer(ContractNode cn, JsonElement arg, JsonObject returnValue) {
ContractRequest onStartingDoipServer = new ContractRequest();
onStartingDoipServer.setAction("onServerStart");
if (arg == null) {
@ -724,49 +717,15 @@ public class ContractProcess {
try {
JsonElement onStartingDoipServerRes = invoke(onStartingDoipServer, funNode).result;
returnValue.add("doipModuleStartResult", onStartingDoipServerRes);
int startPort = ContractProcess.instance.server.getPort() + 1;
if (arg.isJsonObject() && arg.getAsJsonObject().has("doipStartPort")) {
startPort = arg.getAsJsonObject().get("doipStartPort").getAsInt();
}
LOGGER.info("Fetch the onStartingDoipServerRes from router successfully, the result is " + onStartingDoipServerRes);
if(onStartingDoipServerRes.isJsonObject()) {
JsonObject onStartingDoipServerJO = onStartingDoipServerRes.getAsJsonObject();
if (!onStartingDoipServerJO.has("doipAddr")) {
throw new Exception("the doipAddr is improper");
} else {
JsonElement doipAddrJE = onStartingDoipServerJO.get("doipAddr");
if(doipAddrJE.isJsonArray()) {
JsonArray doipAddrJA = doipAddrJE.getAsJsonArray();
for(int i = 0 ; i < doipAddrJA.size() ; i++) {
DoipClusterServer.startDoipServer(doipAddrJA.get(i).getAsString());
}
} else {
DoipClusterServer.startDoipServer(doipAddrJE.getAsString());
}
}
} else {
throw new Exception("the onStartingDoipServerRes doesn't return the correct json result");
}
} catch (Exception e) {
LOGGER.error("DoipLocalSingleton cannot starts properly, plz check the onServerStart function");
e.printStackTrace();
}
}
public void invokeOnStartingDoipServer2(JsonElement arg) {
Object[] funcArgs = new Object[3];
funcArgs[0] = JSONTool.convertJsonElementToMirror(arg);
funcArgs[1] = contract.getOwner();
if (contract.getDoipFlag() && null != contract.getDOI() && !contract.getDOI().isEmpty()) {
funcArgs[2] = contract.getDOI();
} else {
funcArgs[2] = "empty";
}
Object result = engine.invokeFunction("onServerStart", funcArgs);
Map<String, String> resMap = (Map<String, String>) result;
try {
if (!resMap.containsKey("doipAddr")) {
throw new Exception("the doipAddr is improper");
} else {
DoipClusterServer.startDoipServer(resMap.get("doipAddr"));
}
int doipListenPort = DoipClusterServer.startDoipServer(startPort);
returnValue.addProperty("doipListenPort", doipListenPort);
returnValue.addProperty("doipStartPort", startPort);
} catch (Exception e) {
LOGGER.error("DoipLocalSingleton cannot starts properly, plz check the onServerStart function");
e.printStackTrace();

1
src/main/java/org/bdware/sc/handler/DOOPRequestHandler.java
View File

@ -36,7 +36,6 @@ public class DOOPRequestHandler implements DoipRequestHandler, RepositoryHandler
if(instance == null) {
instance = new DOOPRequestHandler();
}
return instance;
}

101
src/main/java/org/bdware/sc/server/DoipClusterServer.java
View File

@ -1,6 +1,5 @@
package org.bdware.sc.server;
import com.google.gson.Gson;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import org.apache.logging.log4j.LogManager;
@ -15,15 +14,10 @@ import org.bdware.irp.exception.IrpClientException;
import org.bdware.irp.stateinfo.StateInfoBase;
import org.bdware.sc.ContractProcess;
import org.bdware.sc.bean.Contract;
import org.bdware.sc.bean.JoinInfo;
import org.bdware.sc.bean.RouteInfo;
import org.bdware.sc.handler.DOOPRequestHandler;
import org.bdware.sc.node.ContractNode;
import org.bdware.sc.node.FunctionNode;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import static org.bdware.doip.audit.EndpointConfig.defaultDOIPServerPort;
import static org.bdware.doip.audit.EndpointConfig.defaultRepoType;
@ -48,21 +42,14 @@ public class DoipClusterServer extends DoipServerImpl {
return instance;
}
public static void startDoipServer(String arg) throws InterruptedException {
final String doipAddr = arg;
Thread doipServerThread = new Thread(){
@Override
public void run() {
try {
DoipLocalSingleton.run(doipAddr);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
doipServerThread.start();
public static int startDoipServer(int startPort) throws InterruptedException {
try {
int ret = DoipLocalSingleton.run(startPort);
return ret;
} catch (Exception e) {
e.printStackTrace();
}
return -1;
}
private static DoipServiceInfo resolveInfo(EndpointConfig config) {
@ -88,7 +75,7 @@ public class DoipClusterServer extends DoipServerImpl {
repoInfo.identifier = repoIdentifier;
repoInfo.handleValues = new JsonObject();
JsonObject repoHandleValues = new JsonObject();
JsonObject createParams = contract.getCreateParam().getAsJsonObject();
@ -104,14 +91,13 @@ public class DoipClusterServer extends DoipServerImpl {
JsonObject functions = new JsonObject();
// 维护RouteInfo，将RouteInfo和doipOperationName的映射关系，以及所有Router中用得到的函数都维护好
maintainRouteJoinInfo(cn, methodRouteInfoMap, methodJoinInfoMap, functions);
if(clusterInfo != null) repoHandleValues.add("clusterInfo", clusterInfo);
if(!functions.equals(new JsonObject())) repoHandleValues.add("functions", functions);
if(!methodRouteInfoMap.equals(new JsonObject())) repoHandleValues.add("routeInfo", methodRouteInfoMap);
if(!methodJoinInfoMap.equals(new JsonObject())) repoHandleValues.add("joinInfo", methodJoinInfoMap);
//TODO 移除这部分逻辑？
cn.maintainRouteJoinInfo(methodRouteInfoMap, methodJoinInfoMap, functions);
if (clusterInfo != null) repoHandleValues.add("clusterInfo", clusterInfo);
if (functions.size() > 0) repoHandleValues.add("functions", functions);
if (methodRouteInfoMap.size() > 0) repoHandleValues.add("routeInfo", methodRouteInfoMap);
if (methodJoinInfoMap.size() > 0) repoHandleValues.add("joinInfo", methodJoinInfoMap);
repoInfo.handleValues.addProperty("cluster", repoHandleValues.toString());
String updateRepoInfoRes = repoIrpClient.reRegister(repoInfo);
if (updateRepoInfoRes.equals("success")) {
LOGGER.info("Update cluster info to router successfully");
@ -122,62 +108,5 @@ public class DoipClusterServer extends DoipServerImpl {
}
}
public void maintainRouteJoinInfo(ContractNode cn, JsonObject methodRouteInfoMap, JsonObject methodJoinInfoMap, JsonObject functions) {
// all functions存了ContractNode中，所有的FunctionNode
List<FunctionNode> allFunctions = cn.getFunctions();
// gson是Gson工具类昂！！！
Gson gson = new Gson();
// 遍历所有的 doipOperationName 和其对应的 doipFunctionNode
for (Map.Entry<String, FunctionNode> doipFunctionNodes : DOOPRequestHandler.instance.doipFunctionNodeMap.entrySet()) {
String doipOperationName = doipFunctionNodes.getKey();
FunctionNode doipFunctionNode = doipFunctionNodes.getValue();
RouteInfo doipFunctionRouteInfo = doipFunctionNode.getRouteInfo();
JoinInfo doipFunctionJoinInfo = doipFunctionNode.getJoinInfo();
// 对于RouterInfo进行维护
if(doipFunctionRouteInfo != null) {
// 建立method和RouteInfo的映射
methodRouteInfoMap.addProperty(doipOperationName, gson.toJson(doipFunctionRouteInfo));
if(doipFunctionRouteInfo.funcName != null) {
String routeFunctionName = doipFunctionRouteInfo.funcName;
packSourceFunctionAndDependentFunctions(allFunctions, routeFunctionName, functions);
}
}
// 对于JoinInfo进行维护
if(doipFunctionJoinInfo != null) {
// 建立method和JoinInfo的映射
methodJoinInfoMap.addProperty(doipOperationName, gson.toJson(doipFunctionJoinInfo));
// 包装JoinInfo中用到的JoinFunction
if(doipFunctionJoinInfo.joinFuncName != null) {
String joinFunctionName = doipFunctionJoinInfo.joinFuncName;
packSourceFunctionAndDependentFunctions(allFunctions, joinFunctionName, functions);
}
// 包装JoinInfo中用到的JoinCountFunction
if(doipFunctionJoinInfo.joinCountFuncName != null) {
String joinFunctionCountName = doipFunctionJoinInfo.joinCountFuncName;
packSourceFunctionAndDependentFunctions(allFunctions, joinFunctionCountName, functions);
}
}
}
}
public void packSourceFunctionAndDependentFunctions(List<FunctionNode> allFunctions, String sourceFunctionName, JsonObject functions) {
for(FunctionNode functionNode : allFunctions) {
// add sourceFunction
if(functionNode.functionName.equals(sourceFunctionName)) {
functions.addProperty(functionNode.functionName, functionNode.plainText());
// find all dependent functions to the "functions" struct
for (String dependentFunctionName : functionNode.getDependentFunctions()) {
for (FunctionNode f : allFunctions) {
if(f.functionName.equals(dependentFunctionName) && !functions.has(dependentFunctionName)) {
functions.addProperty(dependentFunctionName, f.plainText());
}
}
}
}
}
}
}

78
src/main/java/org/bdware/sc/server/DoipLocalSingleton.java
View File

@ -5,19 +5,21 @@ import org.apache.logging.log4j.Logger;
import org.bdware.doip.endpoint.server.DoipListenerConfig;
import org.bdware.doip.endpoint.server.DoipServerImpl;
import org.bdware.doip.endpoint.server.DoipServiceInfo;
import org.bdware.doip.endpoint.server.StartServerCallback;
import org.bdware.sc.handler.DOOPRequestHandler;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
public class DoipLocalSingleton {
static Logger LOGGER = LogManager.getLogger(DoipLocalSingleton.class);
private static DoipServerImpl server;
public static void main(String[] arg) throws InterruptedException {
final int port = (arg.length == 0 ? 21042 : Integer.parseInt(arg[0]));
Thread doipServerThread = new Thread(){
Thread doipServerThread = new Thread() {
@Override
public void run() {
try {
@ -27,45 +29,69 @@ public class DoipLocalSingleton {
}
}
};
doipServerThread.start();
}
public static void run(int port) throws InterruptedException {
List<DoipListenerConfig> infos = new ArrayList<>();
try {
infos.add(new DoipListenerConfig("tcp://127.0.0.1:" + port, "2.1"));
} catch (Exception e) {
e.printStackTrace();
}
DoipServiceInfo info = new DoipServiceInfo("aibd.govdata.tj/do.3f9c41e6-9f8e-48a0-9220-53f438d40e43", "ownerDEF", "gateRepo", infos);
DoipServerImpl server = new DoipServerImpl(info);
final AtomicInteger count = new AtomicInteger(0);
DOOPRequestHandler handler = DOOPRequestHandler.createHandler();
server.setRepositoryHandler(handler);
server.start();
for (; ; ) {
LOGGER.info("Count:" + count.get());
Thread.sleep(10000);
public static int run(int port) throws InterruptedException {
int i = -1;
LOGGER.info("try to listener port:" + port);
for (i = run("tcp://127.0.0.1:" + port++); i < 0; ) {
LOGGER.info("try again to listener port:" + port);
i = run("tcp://127.0.0.1:" + port++);
}
return i;
}
public static void run(String doipAddr) throws InterruptedException {
public static int run(String doipAddr) throws InterruptedException {
List<DoipListenerConfig> infos = new ArrayList<>();
int port = -1;
try {
URI uri = new URI(doipAddr);
port = uri.getPort();
infos.add(new DoipListenerConfig(doipAddr, "2.1"));
} catch (Exception e) {
e.printStackTrace();
}
DoipServiceInfo info = new DoipServiceInfo("aibd.govdata.tj/do.3f9c41e6-9f8e-48a0-9220-53f438d40e43", "ownerDEF", "gateRepo", infos);
DoipServerImpl server = new DoipServerImpl(info);
final AtomicInteger count = new AtomicInteger(0);
server = new DoipServerImpl(info);
DOOPRequestHandler handler = DOOPRequestHandler.createHandler();
server.setRepositoryHandler(handler);
server.start();
for (; ; ) {
LOGGER.info("Count:" + count.get());
Thread.sleep(10000);
ResultChecker checker = new ResultChecker();
server.start(checker);
checker.waitForResult(1000);
if (checker.port > 0)
return port;
return -1;
}
static class ResultChecker implements StartServerCallback {
int port = -2;
@Override
public void onSuccess(int i) {
port = i;
synchronized (this) {
this.notify();
}
}
public void waitForResult(long timeout) {
synchronized (this) {
try {
this.wait(timeout);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
@Override
public void onException(Exception e) {
port = -1;
synchronized (this) {
this.notify();
}
}
}
}