const file = require("./utils/file"); const {getRustConfig} = require("./config"); const {spawn, exec, fork, execSync} = require("child_process"); const {logger} = require("./utils/logger"); const http = require("./utils/http"); const e = require("express"); /******* * web * **** */ const RobotStatus = Object.freeze({ // 已停止的 STOPPED: "STOPPED", // 正在停止 STOP_PENDING: "STOP_PENDING", //正在运行 RUNNING: "RUNNING", //启动中 START_PENDING: "START_PENDING", //下载中 DOWNLOADING: "DOWNLOADING", //错误 ERROR: "ERROR" }); let appMap = new Map(); function getApp(key) { let app = { id: -1, childProcess: undefined, nowBalance: 0, messlist: [], threadStatus: RobotStatus.STOPPED, errorMessage: "成功", threadStartTime: -1, } // logger.info(appMap, appMap.has(key)); if (appMap.has(key)) { app = appMap.get(key) } else { appMap.set(key, app) } // logger.info(app); return app } function delApp(key) { appMap.delete(key) } async function run(param) { var key = param.id var appName = param.path var app = getApp(key) var config = getRustConfig() /**** *** 第一步:防止 反复启动,先要kill一次 ***/ var waitForCondition = function (intervalDuration, conditionFunction) { return new Promise((resolve, reject) => { const intervalId = setInterval(() => { if (conditionFunction()) { clearInterval(intervalId); resolve(); } }, intervalDuration); }); } logger.info('当前app', app); if (app.childProcess !== undefined && app.threadStatus !== RobotStatus.STOPPED && app.threadStatus !== RobotStatus.ERROR && app.threadStatus !== RobotStatus.STOP_PENDING) { logger.info('防止重复启动,需要先杀死'); app.threadStatus = RobotStatus.STOP_PENDING robotStatus(app) app.childProcess.kill() // 当这个条件变为true时,会结束等待 await waitForCondition(1000, () => { if (app.threadStatus === RobotStatus.STOPPED) { logger.info('成功杀死可以重新开启'); return true } else { logger.info('等待杀死'); return false } }); } /********************新杀进程*************************************/ // await waitForCondition(1000, () => { // if (app.threadStatus === RobotStatus.STOPPED) { // logger.info('成功杀死可以重新开启'); // return true // } else { // logger.info('等待杀死'); // return false // } // }); // return; /********************新杀进程*************************************/ // 初始化机器人状态 app.threadStatus = RobotStatus.START_PENDING app.id = key robotStatus(app) /**** *** 第二步:路径经组装 * 注意:可能存在一台服务器多个机器人,通过机器人ID创建文件夹区分,需要组装好路径 ***/ //系统不同 做不同的路径处理 const platform = process.platform; let exeName = appName; //可执行程序 // let configName = "config.toml"; //配置文件 let configName = "config.json"; //配置文件 let appPath = ""; if (platform === 'win32') {//直接运行相对路径 appPath = config.winPath + "/" + app.id } else {//打包为docker 镜像 appPath = config.linuxPath + "/" + app.id } /**** *** 第三步:rust 启动程序检查(下载更新) ***/ var isDow = false var scheduleDow = 0 var scheduleConfig = 0 //1. 检查目录 file.checkPathSync(appPath); //2、 检查执行程序 var destination = appPath + "/" + exeName if (!file.checkFilePath(destination)) { app.threadStatus = RobotStatus.DOWNLOADING robotStatus(app) isDow = true var dowHeaders = {...config.headers}; if (platform === 'win32') { downloadFileWithPowerShell( config.baseUrl + config.dowAppURL + '/?path=' + appName, appPath + "/" + appName, dowHeaders, (err, b) => { if (err === null) { scheduleDow = 1 } else { app.threadStatus = RobotStatus.ERROR app.errorMessage = '下载失败!' scheduleDow = -1 robotStatus(app) } }); } else { downloadFileWithLinux( config.baseUrl + config.dowAppURL + '/?path=' + appName, appPath + "/" + appName, dowHeaders, (err, b) => { if (err === null) { scheduleDow = 1 } else { app.threadStatus = RobotStatus.ERROR app.errorMessage = '下载失败!' scheduleDow = -1 robotStatus(app) } }); } } else { scheduleDow = 1 } /**** *** 第四步:rust 启动配置检查(下载更新) ***/ //2 为防止启动指令不同,每次重新写入 var destination2 = appPath + "/" + configName var urrrl = config.baseUrl + config.dowConfigURL + '/?robotId=' + app.id var configHeaders = {...config.headers}; http.request_get(urrrl, {...config.headers}) .then((data) => { logger.info('配置参数:', data); const json = JSON.parse(data) //处理成配置文件 const map = json.data var json_obj = JSON.parse("{}") for (k in map) { //未完成 参数解析有问题 if (k === "account") { var account = map["account"]; delete map.account json_obj["account_name"] = account["name"] json_obj["access_key"] = account["accessKey"] json_obj["secret_key"] = account["secretKey"] json_obj["pass_key"] = account["pass"] } else if (k === "ref_pair") { json_obj["ref_pair"] = [map["ref_pair"]] } else { json_obj[k] = map[k] } logger.info(map[k] + "\t") } logger.info("参数组装完成!") te = JSON.stringify(json_obj) file.writeFile(destination2, te, (errer, b) => { if (errer === null && b === true) { scheduleConfig = 1 } else { logger.info("配置参数写入配置失败!", errer) app.threadStatus = RobotStatus.ERROR app.errorMessage = '配置参数写入配置失败!' scheduleConfig = -1 robotStatus(app) } }) }) .catch((e) => { logger.info("配置参数获取失败", e) app.threadStatus = RobotStatus.ERROR app.errorMessage = '配置参数获取失败!' scheduleConfig = -1 robotStatus(app) }) //监听下载只有下载完成了才能继续 while (true) { await delay(5000); let info_t = "" let info_t2 = "" //是否开启下载,如果是新下载,下载完成需要授权 if (isDow) { if (scheduleDow === 1) { info_t = "启动文件:下载完成!" //文件授权 execSync(`chmod +x ${appPath + "/" + appName}`, (error, stdout, stderr) => { if (error) { logger.error(`启动文件:授权失败: ${error}`); } logger.info(`启动文件:授权完成!`); }); } else if (scheduleDow === -1) { info_t = "启动文件:下载失败!" } else { info_t = "启动文件:还在下载..." } } else { info_t = "启动文件:完整无需下载" } if (scheduleConfig === 1) { info_t2 = "配置文件:读取成功!" } else if (scheduleConfig === -1) { info_t = "配置文件:读取失败!" } else { info_t2 = "配置文件:还在读取..." } logger.info(info_t, info_t2); if (scheduleDow === 1 && scheduleConfig === 1) { break } else if (scheduleConfig === -1 || scheduleDow === -1) { return; } } app.threadStatus = RobotStatus.START_PENDING robotStatus(app) logger.info("开始启动程序!"); //3. spawn启动 const exePath = appPath + "/" + exeName const configPath = appPath + "/" + configName logger.info(`文件地址:${exePath}-----${configPath}`); const command = exePath const args = ['--config=' + configPath] app.childProcess = spawn(command, args) app.threadStartTime = new Date().getTime() /**********监听*********/ app.childProcess.stdout.on('data', (msg) => { // logger.info('stdout:' + msg.toString()) }) app.childProcess.on('message', (msg) => { // logger.info(`message: ${msg}`); }); app.childProcess.on('exit', (code, signal) => { logger.info(`子进程退出-exit,退出码: ${code}, 信号: ${signal}`); }); app.childProcess.on('close', (code) => { logger.info(`子进程退出-close,退出码 ${code}`); app.threadStatus = RobotStatus.STOPPED robotStatus(app) }); app.threadStatus = RobotStatus.RUNNING } function delay(ms) { return new Promise(resolve => setTimeout(resolve, ms)); } async function closeApp(param) { var key = param.id var app = getApp(key) logger.info(` 信号: `, app.threadStatus); if (app.threadStatus === RobotStatus.RUNNING) { app.childProcess.kill('SIGTERM'); app.threadStatus = RobotStatus.STOP_PENDING robotStatus(app) } } // 下载执行程序,使用的实行服务器指令的方式进行下载 function downloadFileWithPowerShell(url, destination, headers, funBreak) { const headersString = Object.entries(headers) .map(([key, value]) => `'${key}'='${value}'`) // 使用单引号来确保特殊字符不被解析 .join('; '); // 使用分号和空格分隔每个键值对 const command = `powershell -command "Invoke-WebRequest -Uri '${url}' -OutFile '${destination}' -Headers @{${headersString}}"`; exec(command, (error, stdout, stderr) => { if (error) { logger.error(`下载出错: ${error}`); return funBreak(error); } logger.info(`下载完成!`); funBreak(null, true); // 通知下载成功,修改了B为true }); } function downloadFileWithLinux(url, destination, headers, funBreak) { const headersString = Object.entries(headers) .map(([key, value]) => `-H '${key}: ${value}'`) // 设置curl的HTTP头参数 .join(' '); // 使用空格分隔每个头参数 const command = `curl ${headersString} '${url}' -o '${destination}'`; exec(command, (error, stdout, stderr) => { if (error) { logger.error(`下载出错: ${error}`); return funBreak(error); } logger.info(`下载完成!`); funBreak(null, true); // 通知下载成功,修改了B为true }); } // 上报-状态 function robotStatus(app) { var config = getRustConfig() var msg = (app.threadStatus !== RobotStatus.ERROR ? "完成" : app.errorMessage) http.request_post(`${config.baseUrl}/report/statusReport`, { "robotId": app.id, "status": app.threadStatus, "msg": msg }, {...config.headers}).then((data) => { // logger.info('??', data); logger.info('汇报--状态:', '机器人id:', app.id, '状态:', app.threadStatus); }).catch((error) => { logger.error(`请求遇到问题1: ${error.message}`); // 处理可能发生的错误 }); } // 上报-余额 function robotAmount(app) { //拿到策略余额 try { var config = getRustConfig() http.request_get(`${config.rustUrl}/account`, {...config.headers}) .then((data) => { var d = JSON.parse(data) //余额有变动上报一次 if ((app.nowBalance + "") !== (d.now_balance + "")) { http.request_post(`${config.baseUrl}/report/amountReport`, { "robotId": app.id, "amount": d.now_balance }, {...config.headers}) .then((data2) => { // logger.info('上报响应', data2); logger.info('汇报--余额:pid:', app.childProcess.pid, '机器人id:', app.id, '机器人本地余额:', app.nowBalance, '实际余额:', d.now_balance); }).catch((error) => { logger.error(`请求遇到问题:2 ${error.message}`); // 处理可能发生的错误 }); app.nowBalance = d.now_balance } }).catch((error) => { logger.error(`请求遇到问题2: ${error.message}`); // 处理可能发生的错误 }); } catch (e) { logger.error('请求失败!:', e) } } // 定时清理停止状态的 机器人 function delRobot(app) { const platform = process.platform; if (platform !== 'win32' && app.threadStatus === RobotStatus.STOPPED) { var pid = app.childProcess.pid delApp(app.id) } } module.exports = { run, closeApp, delRobot, robotStatus, robotAmount, appMap, RobotStatus };