From 9f30c938dae994d2237c288a4ea70375a923fe98 Mon Sep 17 00:00:00 2001 From: CaiHQ Date: Wed, 29 Dec 2021 11:52:24 +0800 Subject: [PATCH] optimize receive file --- .../server/action/TemporyTestAction.java | 10 +- .../client/NodeCenterClientController.java | 135 ++++++++++++------ 2 files changed, 92 insertions(+), 53 deletions(-) diff --git a/src/main/java/org/bdware/server/action/TemporyTestAction.java b/src/main/java/org/bdware/server/action/TemporyTestAction.java index 0155bd9..d069afa 100644 --- a/src/main/java/org/bdware/server/action/TemporyTestAction.java +++ b/src/main/java/org/bdware/server/action/TemporyTestAction.java @@ -3,10 +3,7 @@ package org.bdware.server.action; import com.google.gson.JsonObject; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; -import org.bdware.sc.ContractClient; -import org.bdware.sc.ContractManager; -import org.bdware.sc.ContractMeta; -import org.bdware.sc.ContractStatusEnum; +import org.bdware.sc.*; import org.bdware.sc.bean.Contract; import org.bdware.sc.bean.ContractExecType; import org.bdware.sc.conn.ResultCallback; @@ -179,8 +176,9 @@ public class TemporyTestAction { } @Action(async = true) - public void reconnectAll(JsonObject args, ResultCallback resultCallback) { - ContractManager.instance.reconnectContractProcess(); + public void reconnectPort(JsonObject args, ResultCallback resultCallback) { + ContractPort.PortVisitor reconnectVisitor = CMActions.manager.statusRecorder.getVisitor(); + reconnectVisitor.visit(args.get("port").getAsInt()); String data = ContractManager.instance.listContractsWithOwner( args.get("owner").getAsString(), null, 0); JsonObject ret = new JsonObject(); diff --git a/src/main/java/org/bdware/server/nodecenter/client/NodeCenterClientController.java b/src/main/java/org/bdware/server/nodecenter/client/NodeCenterClientController.java index 0e3df0e..ddf58fa 100644 --- a/src/main/java/org/bdware/server/nodecenter/client/NodeCenterClientController.java +++ b/src/main/java/org/bdware/server/nodecenter/client/NodeCenterClientController.java @@ -40,7 +40,6 @@ public class NodeCenterClientController implements NodeCenterConn { private static final Logger LOGGER = LogManager.getLogger(NodeCenterClientController.class); public static SyncResult sync = new SyncResult(); private static boolean startCheck = false; - private final Map fileMap; private final NetNeighbors neighbors; public Map distributeReqMap = new ConcurrentHashMap<>(); // public NodeCenterClientController cmClientController; @@ -48,9 +47,11 @@ public class NodeCenterClientController implements NodeCenterConn { NodeCenterClientHandler handler; // 合约contractID，master的公钥 Map contractID2PubKey = new ConcurrentHashMap<>(); + Deque receiveQueue = new ArrayDeque<>(); + ReceiveFileThread receiveFileThread = new ReceiveFileThread(); public NodeCenterClientController(String nodeID) { - this.fileMap = new HashMap<>(); + this.nodeID = nodeID; this.neighbors = new NetNeighbors(); } @@ -518,65 +519,105 @@ public class NodeCenterClientController implements NodeCenterConn { queryNCRepoDOI(json, result); } - @Action(async = true) - public void receiveProject(JsonObject args, final ResultCallback rc) { - LOGGER.debug("position----7"); - String fileName = args.get("fileName").getAsString(); - boolean isAppend = args.get("isAppend").getAsBoolean(); - boolean isDone = args.get("isDone").getAsBoolean(); - boolean isPrivate = args.get("isPrivate").getAsBoolean(); + class ReceiveFileThread extends Thread { + private final Map fileMap = new HashMap<>(); - LOGGER.debug( - String.format("isAppend=%b isDone=%b isPrivate=%b", isAppend, isDone, isPrivate)); - String path = GlobalConf.instance.publicCompiledDir; - if (isPrivate && args.has("pubKey")) { - path = GlobalConf.instance.privateCompiledDir + "/" + args.get("pubKey").getAsString(); + ReceiveFileThread() { + super(); + this.start(); } - File dir = new File(path); - if (!dir.exists()) { - LOGGER.debug("mkdir " + dir.getAbsoluteFile() + ": " + dir.mkdirs()); - } - FileOutputStream fout = null; - if (!isAppend) { - try { - fout = new FileOutputStream(new File(dir, fileName)); - fileMap.put(fileName, fout); - } catch (FileNotFoundException e) { - e.printStackTrace(); + public void run() { + for (; ; ) { + if (receiveQueue.size() > 0) { + try { + JsonObject jo = receiveQueue.poll(); + receiveProject(jo); + } catch (Exception e) { + e.printStackTrace(); + } + } else { + synchronized (ReceiveFileThread.this) { + try { + this.wait(); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + } } - } else { - fout = fileMap.get(fileName); + } - if (isDone) { - if (fout != null) + + public void receiveProject(JsonObject args) { + + String fileName = args.get("fileName").getAsString(); + boolean isAppend = args.get("isAppend").getAsBoolean(); + boolean isDone = args.get("isDone").getAsBoolean(); + boolean isPrivate = args.get("isPrivate").getAsBoolean(); + LOGGER.debug( + String.format("isAppend=%b isDone=%b isPrivate=%b", isAppend, isDone, isPrivate)); + String path = GlobalConf.instance.publicCompiledDir; + if (isPrivate && args.has("pubKey")) { + path = GlobalConf.instance.privateCompiledDir + "/" + args.get("pubKey").getAsString(); + } + File dir = new File(path); + if (!dir.exists()) { + LOGGER.debug("mkdir " + dir.getAbsoluteFile() + ": " + dir.mkdirs()); + } + FileOutputStream fout = null; + if (!isAppend) { try { - fout.close(); - fileMap.remove(fileName); + fout = new FileOutputStream(new File(dir, fileName)); + fileMap.put(fileName, fout); + } catch (FileNotFoundException e) { + e.printStackTrace(); + } + } else { + fout = fileMap.get(fileName); + } + if (isDone) { + if (fout != null) + try { + fout.close(); + fileMap.remove(fileName); + } catch (IOException e) { + e.printStackTrace(); + } + LOGGER.debug("receive finish."); + Map req = new HashMap<>(); + req.put("action", "onReceiveProject"); + req.put("requestID", args.get("requestID").getAsString()); + req.put("nodeID", nodeID); + req.put("progress", "100"); + NetworkManager.instance.sendToNodeCenter(JsonUtil.toJson(req)); + } else { + String data = args.get("data").getAsString(); + byte[] byteData = ByteUtil.decodeBASE64(data); + try { + if (null != fout && null != byteData) { + fout.write(byteData); + } } catch (IOException e) { e.printStackTrace(); } - LOGGER.debug("receive finish."); - Map req = new HashMap<>(); - req.put("action", "onReceiveProject"); - req.put("requestID", args.get("requestID").getAsString()); - req.put("nodeID", nodeID); - req.put("progress", "100"); - rc.onResult(JsonUtil.toJson(req)); - } else { - String data = args.get("data").getAsString(); - byte[] byteData = ByteUtil.decodeBASE64(data); - try { - if (null != fout && null != byteData) { - fout.write(byteData); - } - } catch (IOException e) { - e.printStackTrace(); } } } + @Action(async = false) + public synchronized void receiveProject(JsonObject args, final ResultCallback rc) { + try { + receiveQueue.add(args); + synchronized (receiveFileThread) { + receiveFileThread.notify(); + } + } catch (Exception e) { + e.printStackTrace(); + } + } + public void queryNCRepoDOI(JsonObject json, ResultCallback result) { LOGGER.debug("sendProject: position ---- 3"); String projectName = json.get("projectName").getAsString();