Skip to content

Z
zhichan
Commit 4cf8665c by 蒋勇
Options

d

parent 2509df8f
Showing with 34 additions and 2273 deletions
queue-center/app/config/localsettings.js → center-manage/app/config/localsettings.js 100755 → 100644
var settings = { var settings = {
redis: { redis: {
// host: "121.36.3.35", host: "192.168.4.119",
// port: 6379, port: 6379,
host: "43.247.184.32", password: "Gongsibao2018",
port: 8967, db: 9,
password: "Gongsibao2018", },
db: 5, database: {
}, dbname: "center-manage",
database: { user: "write",
dbname: "tx-queue-log", password: "write",
user: "write", config: {
password: "write", host: '43.247.184.35',
config: { port: 8899,
host: '43.247.184.35', dialect: 'mysql',
port: 8899, timezone:'+08:00',
dialect: 'mysql', operatorsAliases: false,
operatorsAliases: false, pool: {
pool: { max: 5,
max: 5, min: 0,
min: 0, acquire: 90000000,
acquire: 90000000, idle: 1000000
idle: 1000000 },
}, debug: false,
debug: false, dialectOptions: {
dialectOptions: { requestTimeout: 999999,
requestTimeout: 999999, // instanceName:'DEV'
// timezone: '+8:00' } //设置MSSQL超时时间
// instanceName:'DEV' }
} //设置MSSQL超时时间 },
} kongurl: 'http://192.168.4.119:8001/',
}, };
reqEsDevUrl: "http://192.168.4.249:9200/", module.exports = settings;
reqHomePageDevUrl: "http://zcchannel.apps.com:4003/",
reqAuthUrl: "http://sj.app.com:4002/auth" \ No newline at end of file
};
module.exports = settings;
queue-center/app/base/api/api.base.js deleted 100755 → 0
const system = require("../system");
const moment = require('moment');
const settings = require("../../../app/config/settings")
class APIBase {
constructor() {
this.redisClient = system.getObject("util.redisClient");
this.execClient = system.getObject("util.execClient");
this.esUtils = system.getObject("util.esUtils");
this.exTime = 6 * 3600;//缓存过期时间，6小时
}
//-----------------------新的模式------------------开始
async doexecMethod(gname, methodname, pobj, query, req) {
try {
var result = await this[methodname](pobj, query, req);
pobj.actionBody.resultInfo = result;
pobj.actionBody.requestId = result.requestId;
this.esUtils.addEsLogs(settings.queuedName + "api-doexecmethod-info", pobj.actionBody);
return result;
} catch (e) {
console.log(e.stack, "api.base调用出现异常，请联系管理员..........");
var rtnerror = system.getResultFail(-200, "出现异常，error：" + e.stack);
rtnerror.requestId = await this.getBusUid("err");
pobj.actionBody.requestId = rtnerror.requestId;
pobj.actionBody.errorInfo = e.stack;
this.esUtils.addEsLogs(settings.queuedName + "apidoexec-error", pobj.actionBody);
return rtnerror;
}
}
//-----------------------新的模式------------------结束
/**
* 带超时时间的post请求
* @param {*} params 请求数据-json格式
* @param {*} url 请求地址
* @param {*} ContentType 请求头类型，默认application/json
* @param {*} headData 请求头内容-json格式，如：请求头中传递token,格式：{token:"9098902q849q0434q09439"}
*/
async execPostByTimeOut(params, url, ContentType, headData, timeOut = 60) {
var rtn = await this.execClient.execPostTimeOut(params, url, ContentType, headData, timeOut);
if (!rtn || !rtn.stdout) {
return system.getResult(null, "execPostTimeOut data is empty");
}
var result = JSON.parse(rtn.stdout);
return result;
}
/**
* 返回20位业务订单号
* @param {*} prefix 业务前缀
*/
async getBusUid(prefix) {
prefix = (prefix || "");
if (prefix) {
prefix = prefix.toUpperCase();
}
var prefixlength = prefix.length;
var subLen = 8 - prefixlength;
var uidStr = "";
if (subLen > 0) {
uidStr = await this.getUidInfo(subLen, 60);
}
var timStr = moment().format("YYYYMMDDHHmm");
return prefix + timStr + uidStr;
}
/**
* 返回指定长度的字符串
* @param {*} len 返回长度
* @param {*} radix 参与计算的长度，最大为62
*/
async getUidInfo(len, radix) {
var chars = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'.split('');//长度62，到yz长度为长36
var uuid = [], i;
radix = radix || chars.length;
if (len) {
for (i = 0; i < len; i++) uuid[i] = chars[0 | Math.random() * radix];
} else {
var r;
uuid[8] = uuid[13] = uuid[18] = uuid[23] = '-';
uuid[14] = '4';
for (i = 0; i < 36; i++) {
if (!uuid[i]) {
r = 0 | Math.random() * 16;
uuid[i] = chars[(i == 19) ? (r & 0x3) | 0x8 : r];
}
}
}
return uuid.join('');
}
}
module.exports = APIBase;
queue-center/app/base/api/impl/queueAction/producer.js deleted 100755 → 0
var APIBase = require("../../api.base");
var system = require("../../../system");
const settings = require("../../../../config/settings");
class ProducerAPI extends APIBase {
constructor() {
super();
this.utilsProduceSve = system.getObject("service.utilsSve.utilsProduceSve");
}
/**
* 接口跳转-POST请求
* action_type 执行的功能模块
* action_body 执行的参数
*/
async springBoard(pobj, qobj, req) {
if (!pobj.actionType) {
return system.getResult(null, "actionType参数不能为空");
}
var result = await this.opActionType(pobj, pobj.actionType, req);
return result;
}
async opActionType(pobj, actionType, req) {
var opResult = null;
switch (actionType) {
case "produceData":
opResult = await this.utilsProduceSve.produceData(pobj, req);
break;
case "produceLogsData":
if (settings.queuedName != "LOGS-SYTXPUBLIC-MSGQ") {
return system.getResult(null, "请求地址有误!");
}
opResult = await this.utilsProduceSve.produceLogsData(pobj, req);
break;
default:
opResult = system.getResult(null, "actionType参数错误");
break;
}
var uidStr = await this.getBusUid("scz");
opResult.requestId = uidStr;
return opResult;
}
}
module.exports = ProducerAPI;
\ No newline at end of file
queue-center/app/base/db/consumer.base.js deleted 100755 → 0
const system = require("../system");
const moment = require('moment');
class ConsumerBase {
constructor(className) {
this.serviceName = className;
this.execClient = system.getObject("util.execClient");
this.pushSuccessLogDao = system.getObject("db.opLogs.pushSuccessLogDao");
this.pushFailureLogDao = system.getObject("db.opLogs.pushFailureLogDao");
this.errorLogDao = system.getObject("db.opLogs.errorLogDao");
this.redisClient = system.getObject("util.redisClient");
this.duplicateInstance = this.redisClient.getDuplicateInstance();
}
static getServiceName(ClassObj) {
return ClassObj["name"];
}
async doConsumer(queuedName, counter) {
try {
if (!counter) {
counter = 1;
}
var self = this;
this.duplicateInstance.brpop(queuedName, 0, async function (err, repl) {
if (err) {
console.log(err, "......brpop..........err");
return new Error('doConsumer brpop error :' + err);
}
else {
console.log(counter, "......doConsumer.............counter.......................");
if (repl[1]) {
self.execSubDoConsumer(queuedName, JSON.parse(repl[1]));
}
self.doConsumer(queuedName, counter);
}
});
} catch (error) {
this.errorLogDao.addOpErrorLogs(queuedName + "队列执行doConsumer存在异常", null, null, error.stack, 3);
//日志记录
console.log(JSON.stringify({
optitle: this.serviceName + "，队列执行doConsumer存在异常",
op: "base/db/consumer.base.js",
message: ""
}));
}
}
async execSubDoConsumer(queuedName, actionBody) {
var execResult = null;
try {
this.subBeforeConsumer(queuedName, actionBody);
actionBody.requestId = await this.getBusUid("PUB-");
if (queuedName.indexOf("SYTXFAIL-SYTXPUBLIC-MSGQ") < 0) {
execResult = await this.subDoConsumer(queuedName, actionBody);
console.log(execResult, "..........................................11......");
if (execResult.status === 1) {
var notifyQueuedName = "NOTIFY-SYTXPUBLIC-MSGQ";
if (queuedName.indexOf(notifyQueuedName) < 0) {
if (actionBody.notifyUrl && actionBody.notifyUrl.indexOf("http") > 0) {
actionBody.resultInfo = execResult;
console.log(actionBody, "..............................22.......");
await this.redisClient.lpushData(notifyQueuedName, actionBody);
}
}
console.log(actionBody, "........................................33.......");
await this.pushSuccessLogDao.addOpSuccessLogs("推送成功", actionBody, execResult);
return;
}
var failQueuedName = "SYTXFAIL-SYTXPUBLIC-MSGQ";
actionBody.queuedName = queuedName;
actionBody.counter = actionBody.counter ? actionBody.counter : 1;
console.log(actionBody, "...........................................44.......");
await this.redisClient.lpushData(failQueuedName, actionBody);
return;
}
if (actionBody.counter > 4) {
await this.pushFailureLogDao.addOpFailureLogs("推送失败", actionBody, execResult);
return;
}
execResult = await this.subDoConsumer(queuedName, actionBody);
} catch (error) {
this.errorLogDao.addOpErrorLogs(queuedName + "队列执行execSubDoConsumer存在异常", actionBody, execResult, error.stack, 3);
//日志记录
console.log(JSON.stringify({
optitle: this.serviceName + "，队列执行execSubDoConsumer存在异常",
op: "base/db/consumer.base.js",
message: ""
}));
}
}
async subBeforeConsumer(queuedName, actionBody) {
console.log("请在子类中重写此方法进行前置操作......", this.serviceName);
}
async subDoConsumer(queuedName, actionBody) {
throw new Error("请在子类中重写此方法进行操作业务逻辑............................!");
}
sleep(milliSeconds) {
var startTime = new Date().getTime();
while (new Date().getTime() < startTime + milliSeconds);
}
/**
* 带超时时间的post请求
* @param {*} params 请求数据-json格式
* @param {*} url 请求地址
* @param {*} ContentType 请求头类型，默认application/json
* @param {*} headData 请求头内容-json格式，如：请求头中传递token,格式：{token:"9098902q849q0434q09439"}
*/
async execPostByTimeOut(params, url, ContentType, headData, timeOut = 60) {
var rtn = await this.execClient.execPostTimeOut(params, url, ContentType, headData, timeOut);
if (!rtn || !rtn.stdout) {
return system.getResult(null, "execPostTimeOut data is empty");
}
var result = JSON.parse(rtn.stdout);
return result;
}
/*
返回20位业务订单号
prefix：业务前缀
*/
async getBusUid(prefix) {
prefix = (prefix || "");
if (prefix) {
prefix = prefix.toUpperCase();
}
var prefixlength = prefix.length;
var subLen = 8 - prefixlength;
var uidStr = "";
if (subLen > 0) {
uidStr = await this.getUidInfo(subLen, 60);
}
var timStr = moment().format("YYYYMMDDHHmm");
return prefix + timStr + uidStr;
}
/*
len：返回长度
radix：参与计算的长度，最大为62
*/
async getUidInfo(len, radix) {
var chars = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'.split('');//长度62，到yz长度为长36
var uuid = [], i;
radix = radix || chars.length;
if (len) {
for (i = 0; i < len; i++) uuid[i] = chars[0 | Math.random() * radix];
} else {
var r;
uuid[8] = uuid[13] = uuid[18] = uuid[23] = '-';
uuid[14] = '4';
for (i = 0; i < 36; i++) {
if (!uuid[i]) {
r = 0 | Math.random() * 16;
uuid[i] = chars[(i == 19) ? (r & 0x3) | 0x8 : r];
}
}
}
return uuid.join('');
}
}
module.exports = ConsumerBase;
queue-center/app/base/db/consumer/publicFail/publicFailConsumer.js deleted 100755 → 0
const ConsumerBase = require("../../consumer.base");
const system = require("../../../system");
const moment = require('moment');
class PublicFailConsumer extends ConsumerBase {
constructor() {
super(ConsumerBase.getServiceName(PublicFailConsumer));
}
async subBeforeConsumer(queuedName, actionBody) {
console.log("前置操作......", this.serviceName);
}
async subDoConsumer(queuedName, actionBody) {
console.log(actionBody, "......publicFailConsumer.....");
actionBody.counter = actionBody.counter + 1;
var exTime = actionBody.counter * 10;
var mathStr = await this.getUidInfo(3, 10);
var scoreValue = Number(moment().format("YYYYMMDDHHmmssSSS") + mathStr);
await this.redisClient.setDelayEventData(queuedName, scoreValue, exTime);
await this.redisClient.zaddSortedSet(queuedName, scoreValue, actionBody);
return system.getResultSuccess();
}
}
module.exports = PublicFailConsumer;
\ No newline at end of file
queue-center/app/base/db/consumer/publicLogs/publicLogsConsumer.js deleted 100755 → 0
const ConsumerBase = require("../../consumer.base");
const system = require("../../../system");
const moment = require('moment');
class PublicLogsConsumer extends ConsumerBase {
constructor() {
super(ConsumerBase.getServiceName(PublicLogsConsumer));
this.configInfoDao = system.getObject("db.opLogs.configInfoDao");
this.errorLogDao = system.getObject("db.opLogs.errorLogDao");
this.esUtils = system.getObject("util.esUtils");
}
async subBeforeConsumer(queuedName, actionBody) {
console.log("前置操作......", this.serviceName);
}
async subDoConsumer(queuedName, actionBody) {
console.log(actionBody, "......publicLogsConsumer.....");
var params = {
opTitle: moment().format("YYYY-MM-DD-HH-mm-ss") + actionBody.opTitle,
identifyCode: actionBody.identifyCode,
messageBody: actionBody.messageBody,
requestId: actionBody.requestId
}
var execResult = await this.esUtils.addEsLogs(queuedName, actionBody);;
return execResult;
}
}
module.exports = PublicLogsConsumer;
\ No newline at end of file
queue-center/app/base/db/consumer/publicNotify/publicNotifyConsumer.js deleted 100755 → 0
const ConsumerBase = require("../../consumer.base");
class PublicNotifyConsumer extends ConsumerBase {
constructor() {
super(ConsumerBase.getServiceName(PublicNotifyConsumer));
}
async subBeforeConsumer(queuedName, actionBody) {
console.log("前置操作......", this.serviceName);
}
async subDoConsumer(queuedName, actionBody) {
console.log(actionBody, "......PublicNotifyConsumer.....");
var params = {
actionType: actionBody.actionType,
identifyCode: actionBody.identifyCode,
actionBody: {
messageBody: actionBody.messageBody,
resultInfo: actionBody.resultInfo
},
requestId: actionBody.requestId
}
var execResult = await this.execPostByTimeOut(params, actionBody.notifyUrl);
return execResult;
}
}
module.exports = PublicNotifyConsumer;
\ No newline at end of file
queue-center/app/base/db/consumer/publicServiceAllocation/publicConsumer.js deleted 100755 → 0
const ConsumerBase = require("../../consumer.base");
class PublicConsumer extends ConsumerBase {
constructor() {
super(ConsumerBase.getServiceName(PublicConsumer));
}
async subBeforeConsumer(queuedName, actionBody) {
console.log("前置操作......", this.serviceName);
}
async subDoConsumer(queuedName, actionBody) {
console.log(actionBody, "......PublicConsumer.....");
var params = {
actionType: actionBody.actionType,
identifyCode: actionBody.identifyCode,
actionBody: actionBody.messageBody,
requestId: actionBody.requestId
}
var execResult = await this.execPostByTimeOut(params, actionBody.pushUrl);
return execResult;
}
}
module.exports = PublicConsumer;
\ No newline at end of file
queue-center/app/base/db/dao.base.js deleted 100755 → 0
const system = require("../system");
class Dao {
constructor(modelName) {
this.modelName = modelName;
var db = system.getObject("db.common.connection").getCon();
this.db = db;
this.model = db.models[this.modelName];
}
static getModelName(ClassObj) {
var nameStr = ClassObj["name"].substring(0, ClassObj["name"].lastIndexOf("Dao"));
var initialStr = nameStr.substring(0, 1);
var resultStr = initialStr.toLowerCase() + nameStr.substring(1, nameStr.length);
return resultStr;
}
preCreate(u) {
return u;
}
/**
*
* @param {*} u 对象
* @param {*} t 事务对象t
*/
async create(u, t) {
var u2 = this.preCreate(u);
if (t) {
return this.model.create(u2, { transaction: t }).then(u => {
return u;
});
} else {
return this.model.create(u2, { transaction: t }).then(u => {
return u;
});
}
}
async customQuery(sql, paras, t) {
var tmpParas = null;//||paras=='undefined'?{type: this.db.QueryTypes.SELECT }:{ replacements: paras, type: this.db.QueryTypes.SELECT };
if (t && t != 'undefined') {
if (paras == null || paras == 'undefined') {
tmpParas = { type: this.db.QueryTypes.SELECT };
tmpParas.transaction = t;
} else {
tmpParas = { replacements: paras, type: this.db.QueryTypes.SELECT };
tmpParas.transaction = t;
}
} else {
tmpParas = paras == null || paras == 'undefined' || paras.keys == 0 ? { type: this.db.QueryTypes.SELECT } : { replacements: paras, type: this.db.QueryTypes.SELECT };
}
var result = this.db.query(sql, tmpParas);
return result;
}
}
module.exports = Dao;
queue-center/app/base/db/impl/common/connection.js deleted 100755 → 0
const Sequelize = require('sequelize');
const settings=require("../../../../config/settings")
const fs=require("fs")
const path=require("path");
var glob = require("glob");
class DbFactory{
constructor(){
const dbConfig=settings.database();
this.db=new Sequelize(dbConfig.dbname,
dbConfig.user,
dbConfig.password,
dbConfig.config);
this.db.Sequelize=Sequelize;
this.db.Op=Sequelize.Op;
this.initModels();
this.initRelations();
}
async initModels(){
var self=this;
var modelpath=path.normalize(path.join(__dirname, '../..'))+"/models/";
console.log(modelpath);
var models=glob.sync(modelpath+"/**/*.js");
console.log(models.length);
models.forEach(function(m){
self.db.import(m);
});
console.log("init models....");
}
async initRelations(){
}
//async getCon(){,用于使用替换table模型内字段数据使用
getCon(){
var that=this;
// await this.db.authenticate().then(()=>{
// console.log('Connection has been established successfully.');
// }).catch(err => {
// console.error('Unable to connect to the database:', err);
// throw err;
// });
//同步模型
if(settings.env=="dev"){
//console.log(pa);
// pconfigObjs.forEach(p=>{
// console.log(p.get({plain:true}));
// });
// await this.db.models.user.create({nickName:"dev","description":"test user",openId:"testopenid",unionId:"testunionid"})
// .then(function(user){
// var acc=that.db.models.account.build({unionId:"testunionid",nickName:"dev"});
// acc.save().then(a=>{
// user.setAccount(a);
// });
// });
}
return this.db;
}
}
module.exports=DbFactory;
// const dbf=new DbFactory();
// dbf.getCon().then((db)=>{
// //console.log(db);
// // db.models.user.create({nickName:"jy","description":"cccc",openId:"xxyy",unionId:"zz"})
// // .then(function(user){
// // var acc=db.models.account.build({unionId:"zz",nickName:"jy"});
// // acc.save().then(a=>{
// // user.setAccount(a);
// // });
// // console.log(user);
// // });
// // db.models.user.findAll().then(function(rs){
// // console.log("xxxxyyyyyyyyyyyyyyyyy");
// // console.log(rs);
// // })
// });
// const User = db.define('user', {
// firstName: {
// type: Sequelize.STRING
// },
// lastName: {
// type: Sequelize.STRING
// }
// });
// db
// .authenticate()
// .then(() => {
// console.log('Co+nnection has been established successfully.');
//
// User.sync(/*{force: true}*/).then(() => {
// // Table created
// return User.create({
// firstName: 'John',
// lastName: 'Hancock'
// });
// });
//
// })
// .catch(err => {
// console.error('Unable to connect to the database:', err);
// });
//
// User.findAll().then((rows)=>{
// console.log(rows[0].firstName);
// });
queue-center/app/base/db/impl/opLogs/configInfoDao.js deleted 100755 → 0
const system = require("../../../system");
const Dao = require("../../dao.base");
class ConfigInfoDao extends Dao {
constructor() {
super(Dao.getModelName(ConfigInfoDao));
this.configInfoCacheKey = "queue-configuration:info";
this.redisClient = system.getObject("util.redisClient");
}
/**
* 获取配置信息列表，缓存600s
*/
async getList() {
var cacheStr = await this.redisClient.getCache(this.configInfoCacheKey);
if (cacheStr && cacheStr != "undefined") {
var list = JSON.parse(cacheStr);
return system.getResultSuccess(list);
}
var list = await this.model.findAll({
where: { is_enabled: 1 },
raw: true,
attributes: [
"c_key",
"c_value"
]
});
if (!list || list.length === 0) {
return system.getResult(null, "data is empty!");
}
cacheStr = JSON.stringify(list);
this.redisClient.setWithEx(this.configInfoCacheKey, cacheStr, 600);
var result = await system.getResultSuccess(list);
return result;
}
}
module.exports = ConfigInfoDao;
queue-center/app/base/db/impl/opLogs/errorLogDao.js deleted 100755 → 0
const system = require("../../../system");
const Dao = require("../../dao.base");
class ErrorLogDao extends Dao {
constructor() {
super(Dao.getModelName(ErrorLogDao));
}
/**
* 添加错误日志
* @param {*} opTitle 操作标题
* @param {*} actionBody 操作的消息体
* @param {*} execResult 操作的返回信息
* @param {*} error 错误信息
* @param {*} logLevel 日志级别信息,debug: 0, info: 1, warn: 2, error: 3, fatal: 4
*/
async addOpErrorLogs(opTitle, actionBody, execResult, error,logLevel) {
error = typeof error === 'object' ? error : { error_info: error };
var params = {
identify_code: actionBody.identifyCode,
op_title: opTitle,
push_content: actionBody || null,
error_info: error || null,
result_info: execResult || null,
request_id: actionBody.requestId,
client_ip: actionBody.clientIp || "",
created_at: new Date()
}
await this.create(params);
return system.getResultSuccess();
}
}
module.exports = ErrorLogDao;
queue-center/app/base/db/impl/opLogs/pushFailureLogDao.js deleted 100755 → 0
const system = require("../../../system");
const Dao = require("../../dao.base");
class PushFailureLogDao extends Dao {
constructor() {
super(Dao.getModelName(PushFailureLogDao));
}
/**
* 添加失败日志
* @param {*} opTitle 操作标题
* @param {*} actionBody 操作的消息体
* @param {*} execResult 操作的返回信息
*/
async addOpFailureLogs(opTitle, actionBody, execResult) {
var params = {
identify_code: actionBody.identifyCode,
op_title: opTitle,
push_content: actionBody,
result_info: execResult,
request_id: actionBody.requestId,
client_ip: actionBody.clientIp || "",
created_at: new Date()
}
await this.create(params);
return system.getResultSuccess();
}
}
module.exports = PushFailureLogDao;
queue-center/app/base/db/impl/opLogs/pushSuccessLogDao.js deleted 100755 → 0
const system = require("../../../system");
const Dao = require("../../dao.base");
class PushSuccessLogDao extends Dao {
constructor() {
super(Dao.getModelName(PushSuccessLogDao));
}
/**
* 添加成功日志
* @param {*} opTitle 操作标题
* @param {*} actionBody 操作的消息体
* @param {*} execResult 操作的返回信息
*/
async addOpSuccessLogs(opTitle, actionBody, execResult) {
var params = {
identify_code: actionBody.identifyCode,
op_title: opTitle,
push_content: actionBody,
result_info: execResult,
request_id: actionBody.requestId,
client_ip: actionBody.clientIp || "",
created_at: new Date()
}
await this.create(params);
return system.getResultSuccess();
}
}
module.exports = PushSuccessLogDao;
queue-center/app/base/db/models/opLogs/configInfo.js deleted 100755 → 0
const system = require("../../../system");
module.exports = (db, DataTypes) => {
return db.define("configInfo", {
c_key: DataTypes.STRING(100),
c_value: DataTypes.STRING(100),
is_enabled: DataTypes.INTEGER,
}, {
paranoid: false,//假的删除
underscored: true,
version: true,
freezeTableName: true,
timestamps: true,
updatedAt: false,
//freezeTableName: true,
// define the table's name
tableName: 'configuration_info',
validate: {
},
indexes: [
// Create a unique index on email
// {
// unique: true,
// fields: ['email']
// },
//
// // Creates a gin index on data with the jsonb_path_ops operator
// {
// fields: ['data'],
// using: 'gin',
// operator: 'jsonb_path_ops'
// },
//
// // By default index name will be [table]_[fields]
// // Creates a multi column partial index
// {
// name: 'public_by_author',
// fields: ['author', 'status'],
// where: {
// status: 'public'
// }
// },
//
// // A BTREE index with a ordered field
// {
// name: 'title_index',
// method: 'BTREE',
// fields: ['author', {attribute: 'title', collate: 'en_US', order: 'DESC', length: 5}]
// }
]
});
}
queue-center/app/base/db/models/opLogs/errorLog.js deleted 100755 → 0
const system = require("../../../system");
module.exports = (db, DataTypes) => {
return db.define("errorLog", {
identify_code: DataTypes.STRING(100), //标识code
op_title: DataTypes.STRING(100), // 操作标题
push_content: DataTypes.JSON, //推送的内容
error_info: DataTypes.JSON, //错误内容
result_info: DataTypes.JSON, //返回信息
request_id: DataTypes.STRING(100),
client_ip: DataTypes.STRING(100), //请求ip
log_level: DataTypes.INTEGER,// debug: 0, info: 1, warn: 2, error: 3, fatal: 4
}, {
paranoid: false,//假的删除
underscored: true,
version: true,
freezeTableName: true,
timestamps: true,
updatedAt: false,
//freezeTableName: true,
// define the table's name
tableName: 'op_error_log',
validate: {
},
indexes: [
// Create a unique index on email
// {
// unique: true,
// fields: ['email']
// },
//
// // Creates a gin index on data with the jsonb_path_ops operator
// {
// fields: ['data'],
// using: 'gin',
// operator: 'jsonb_path_ops'
// },
//
// // By default index name will be [table]_[fields]
// // Creates a multi column partial index
// {
// name: 'public_by_author',
// fields: ['author', 'status'],
// where: {
// status: 'public'
// }
// },
//
// // A BTREE index with a ordered field
// {
// name: 'title_index',
// method: 'BTREE',
// fields: ['author', {attribute: 'title', collate: 'en_US', order: 'DESC', length: 5}]
// }
]
});
}
queue-center/app/base/db/models/opLogs/pushFailureLog.js deleted 100755 → 0
const system = require("../../../system");
module.exports = (db, DataTypes) => {
return db.define("pushFailureLog", {
identify_code: DataTypes.STRING(100), //标识code
op_title: DataTypes.STRING(100), // 操作标题
push_content: DataTypes.JSON, //推送的内容
result_info: DataTypes.JSON, //返回信息
request_id: DataTypes.STRING(100),
client_ip: DataTypes.STRING(100), //请求ip
}, {
paranoid: false,//假的删除
underscored: true,
version: true,
freezeTableName: true,
timestamps: true,
updatedAt: false,
//freezeTableName: true,
// define the table's name
tableName: 'x_push_failure_log',
validate: {
},
indexes: [
// Create a unique index on email
// {
// unique: true,
// fields: ['email']
// },
//
// // Creates a gin index on data with the jsonb_path_ops operator
// {
// fields: ['data'],
// using: 'gin',
// operator: 'jsonb_path_ops'
// },
//
// // By default index name will be [table]_[fields]
// // Creates a multi column partial index
// {
// name: 'public_by_author',
// fields: ['author', 'status'],
// where: {
// status: 'public'
// }
// },
//
// // A BTREE index with a ordered field
// {
// name: 'title_index',
// method: 'BTREE',
// fields: ['author', {attribute: 'title', collate: 'en_US', order: 'DESC', length: 5}]
// }
]
});
}
queue-center/app/base/db/models/opLogs/pushSuccessLog.js deleted 100755 → 0
const system = require("../../../system");
module.exports = (db, DataTypes) => {
return db.define("pushSuccessLog", {
identify_code: DataTypes.STRING(100), //标识code
op_title: DataTypes.STRING(100), // 操作标题
push_content: DataTypes.JSON, //推送的内容
result_info: DataTypes.JSON, //返回信息
request_id: DataTypes.STRING(100),
client_ip: DataTypes.STRING(100), //请求ip
}, {
paranoid: false,//假的删除
underscored: true,
version: true,
freezeTableName: true,
timestamps: true,
updatedAt: false,
//freezeTableName: true,
// define the table's name
tableName: 'x_push_success_log',
validate: {
},
indexes: [
// Create a unique index on email
// {
// unique: true,
// fields: ['email']
// },
//
// // Creates a gin index on data with the jsonb_path_ops operator
// {
// fields: ['data'],
// using: 'gin',
// operator: 'jsonb_path_ops'
// },
//
// // By default index name will be [table]_[fields]
// // Creates a multi column partial index
// {
// name: 'public_by_author',
// fields: ['author', 'status'],
// where: {
// status: 'public'
// }
// },
//
// // A BTREE index with a ordered field
// {
// name: 'title_index',
// method: 'BTREE',
// fields: ['author', {attribute: 'title', collate: 'en_US', order: 'DESC', length: 5}]
// }
]
});
}
queue-center/app/base/service/app.base.js deleted 100755 → 0
const system = require("../system");
const moment = require('moment')
const settings = require("../../config/settings");
class AppServiceBase {
constructor() {
this.execClient = system.getObject("util.execClient");
this.redisClient = system.getObject("util.redisClient");
}
/**
* 带超时时间的post请求
* @param {*} params 请求数据-json格式
* @param {*} url 请求地址
* @param {*} ContentType 请求头类型，默认application/json
* @param {*} headData 请求头内容-json格式，如：请求头中传递token,格式：{token:"9098902q849q0434q09439"}
*/
async execPostByTimeOut(params, url, ContentType, headData, timeOut = 60) {
var rtn = await this.execClient.execPostTimeOut(params, url, ContentType, headData, timeOut);
if (!rtn || !rtn.stdout) {
return system.getResult(null, "execPostTimeOut data is empty");
}
var result = JSON.parse(rtn.stdout);
return result;
}
async getConvertSemiangleStr(str) {//去除空格及全角转半角
var result = "";
str = str.replace(/\s+/g, "");
var len = str.length;
for (var i = 0; i < len; i++) {
var cCode = str.charCodeAt(i);
//全角与半角相差（除空格外）：65248（十进制）
cCode = (cCode >= 0xFF01 && cCode <= 0xFF5E) ? (cCode - 65248) : cCode;
//处理空格
cCode = (cCode == 0x03000) ? 0x0020 : cCode;
result += String.fromCharCode(cCode);
}
return result;
}
/*
返回20位业务订单号
prefix：业务前缀
*/
async getBusUid(prefix) {
prefix = (prefix || "");
if (prefix) {
prefix = prefix.toUpperCase();
}
var prefixlength = prefix.length;
var subLen = 8 - prefixlength;
var uidStr = "";
if (subLen > 0) {
uidStr = await this.getUidInfo(subLen, 60);
}
var timStr = moment().format("YYYYMMDDHHmm");
return prefix + timStr + uidStr;
}
/*
len：返回长度
radix：参与计算的长度，最大为62
*/
async getUidInfo(len, radix) {
var chars = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'.split('');//长度62，到yz长度为长36
var uuid = [], i;
radix = radix || chars.length;
if (len) {
for (i = 0; i < len; i++) uuid[i] = chars[0 | Math.random() * radix];
} else {
var r;
uuid[8] = uuid[13] = uuid[18] = uuid[23] = '-';
uuid[14] = '4';
for (i = 0; i < 36; i++) {
if (!uuid[i]) {
r = 0 | Math.random() * 16;
uuid[i] = chars[(i == 19) ? (r & 0x3) | 0x8 : r];
}
}
}
return uuid.join('');
}
}
module.exports = AppServiceBase;
queue-center/app/base/service/impl/opLogs/configInfoSve.js deleted 100755 → 0
const system = require("../../../system");
const ServiceBase = require("../../sve.base");
class ConfigInfoService extends ServiceBase {
constructor() {
super("opLogs", ServiceBase.getDaoName(ConfigInfoService));
}
}
module.exports = ConfigInfoService;
queue-center/app/base/service/impl/opLogs/errorLogSve.js deleted 100755 → 0
const system = require("../../../system");
const ServiceBase = require("../../sve.base");
class ErrorLogService extends ServiceBase {
constructor() {
super("opLogs", ServiceBase.getDaoName(ErrorLogService));
}
}
module.exports = ErrorLogService;
queue-center/app/base/service/impl/opLogs/pushFailureLogSve.js deleted 100755 → 0
const system = require("../../../system");
const ServiceBase = require("../../sve.base");
class PushFailureLogService extends ServiceBase {
constructor() {
super("opLogs", ServiceBase.getDaoName(PushFailureLogService));
}
}
module.exports = PushFailureLogService;
queue-center/app/base/service/impl/opLogs/pushSuccessLogSve.js deleted 100755 → 0
const system = require("../../../system");
const ServiceBase = require("../../sve.base");
class PushSuccessLogService extends ServiceBase {
constructor() {
super("opLogs", ServiceBase.getDaoName(PushSuccessLogService));
}
}
module.exports = PushSuccessLogService;
queue-center/app/base/service/impl/utilsSve/utilsProduceSve.js deleted 100755 → 0
const system = require("../../../system");
const settings = require("../../../../config/settings");
const AppServiceBase = require("../../app.base");
//用户权限操作
class UtilsProduceService extends AppServiceBase {
constructor() {
super();
}
/**
* 接口跳转-POST请求
* action_type 执行的类型
* action_body 执行的参数
*/
async produceData(pobj, req) {
if (!pobj.actionBody) {
return system.getResult(null, "actionBody不能为空");
}
var keyCount = Object.keys(pobj.actionBody).length;
if (keyCount === 0) {
return system.getResult(null, "actionBody参数不能为空");
}
if (!pobj.actionBody.pushUrl) {
return system.getResult(null, "actionBody.pushUrl参数不能为空");
}
if (!pobj.actionBody.identifyCode) {
return system.getResult(null, "actionBody.identifyCode参数不能为空");
}
if (!pobj.actionBody.messageBody) {
return system.getResult(null, "actionBody.messageBody不能为空");
}
var msgKeyCount = Object.keys(pobj.actionBody.messageBody).length;
if (msgKeyCount === 0) {
return system.getResult(null, "actionBody.messageBody参数不能为空");
}
await this.redisClient.lpushData(settings.queuedName, pobj.actionBody);
return system.getResultSuccess();
}
/**
* 接口跳转-POST请求
* action_type 执行的类型
* action_body 执行的参数
*/
async produceLogsData(pobj, req) {
if (!pobj.actionBody) {
return system.getResult(null, "actionBody不能为空");
}
var keyCount = Object.keys(pobj.actionBody).length;
if (keyCount === 0) {
return system.getResult(null, "actionBody参数不能为空");
}
if (!pobj.actionBody.identifyCode) {
return system.getResult(null, "actionBody.identifyCode参数不能为空");
}
if (!pobj.actionBody.messageBody) {
return system.getResult(null, "actionBody.messageBody不能为空");
}
var msgKeyCount = Object.keys(pobj.actionBody.messageBody).length;
if (msgKeyCount === 0) {
return system.getResult(null, "actionBody.messageBody参数不能为空");
}
await this.redisClient.lpushData(settings.queuedName, pobj.actionBody);
return system.getResultSuccess();
}
}
module.exports = UtilsProduceService;
queue-center/app/base/service/sve.base.js deleted 100755 → 0
const system = require("../system");
const moment = require('moment')
// const settings = require("../../config/settings");
class ServiceBase {
constructor(gname, daoName) {
this.execClient = system.getObject("util.execClient");
this.db = system.getObject("db.common.connection").getCon();
this.daoName = daoName;
this.dao = system.getObject("db." + gname + "." + daoName);
}
static getDaoName(ClassObj) {
var nameStr = ClassObj["name"].substring(0, ClassObj["name"].lastIndexOf("Service")) + "Dao";
var initialStr = nameStr.substring(0, 1);
var resultStr = initialStr.toLowerCase() + nameStr.substring(1, nameStr.length);
return resultStr;
}
/**
* 带超时时间的post请求
* @param {*} params 请求数据-json格式
* @param {*} url 请求地址
* @param {*} ContentType 请求头类型，默认application/json
* @param {*} headData 请求头内容-json格式，如：请求头中传递token,格式：{token:"9098902q849q0434q09439"}
*/
async execPostByTimeOut(params, url, ContentType, headData, timeOut = 60) {
var rtn = await this.execClient.execPostTimeOut(params, url, ContentType, headData, timeOut);
if (!rtn || !rtn.stdout) {
return system.getResult(null, "execPostTimeOut data is empty");
}
var result = JSON.parse(rtn.stdout);
return result;
}
async create(qobj) {
return this.dao.create(qobj);
}
async customQuery(sql, paras, t) {
return this.dao.customQuery(sql, paras, t);
}
/*
返回20位业务订单号
prefix：业务前缀
*/
async getBusUid(prefix) {
prefix = (prefix || "");
if (prefix) {
prefix = prefix.toUpperCase();
}
var prefixlength = prefix.length;
var subLen = 8 - prefixlength;
var uidStr = "";
if (subLen > 0) {
uidStr = await this.getUidInfo(subLen, 60);
}
var timStr = moment().format("YYYYMMDDHHmm");
return prefix + timStr + uidStr;
}
/*
len：返回长度
radix：参与计算的长度，最大为62
*/
async getUidInfo(len, radix) {
var chars = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'.split('');//长度62，到yz长度为长36
var uuid = [], i;
radix = radix || chars.length;
if (len) {
for (i = 0; i < len; i++) uuid[i] = chars[0 | Math.random() * radix];
} else {
var r;
uuid[8] = uuid[13] = uuid[18] = uuid[23] = '-';
uuid[14] = '4';
for (i = 0; i < 36; i++) {
if (!uuid[i]) {
r = 0 | Math.random() * 16;
uuid[i] = chars[(i == 19) ? (r & 0x3) | 0x8 : r];
}
}
}
return uuid.join('');
}
}
module.exports = ServiceBase;
queue-center/app/base/system.js deleted 100755 → 0
var fs = require("fs");
var objsettings = require("../config/objsettings");
var settings = require("../config/settings");
const request = require('request');
class System {
static declare(ns) {
var ar = ns.split('.');
var root = System;
for (var i = 0, len = ar.length; i < len; ++i) {
var n = ar[i];
if (!root[n]) {
root[n] = {};
root = root[n];
} else {
root = root[n];
}
}
}
static async delReq(url, qdata) {
let rtn = {}
let promise = new Promise(function (resv, rej) {
request.del({
url: url,
qs: qdata
}, function (error, response, body) {
rtn.statusCode = response.statusCode
if (!error) {
if (body) {
let data = JSON.parse(body)
rtn.data = data
} else {
rtn.data = null
}
resv(rtn);
} else {
rej(error)
}
});
})
return promise;
}
static async getReq(url, qdata) {
let rtn = {}
let promise = new Promise(function (resv, rej) {
request.get({
url: url,
json: true,
qs: qdata
}, function (error, response, body) {
rtn.statusCode = response.statusCode;
if (!error) {
if (body) {
rtn.data = body
} else {
rtn.data = null
}
resv(rtn);
} else {
rej(error);
}
});
})
return promise;
}
static async postJsonTypeReq(url, data, md = "POST") {
let rtn = {}
let promise = new Promise(function (resv, rej) {
request({
url: url,
method: md,
json: true,
headers: {
"content-type": "application/json",
},
body: data
}, function (error, response, body) {
console.log(error)
rtn.statusCode = response.statusCode
if (!error) {
if (body) {
rtn.data = body
} else {
rtn.data = null
}
resv(rtn);
} else {
rej(error)
}
});
})
return promise;
}
static async post3wFormTypeReq(url, data) {
let rtn = {}
let promise = new Promise(function (resv, rej) {
request.post({
url: url,
form: data
}, function (error, response, body) {
rtn.statusCode = response.statusCode
if (!error) {
let data = JSON.parse(body)
rtn.data = data
resv(rtn);
} else {
rej(error)
}
});
})
return promise;
}
static async postMpFormTypeReq(url, formdata) {
let promise = new Promise(function (resv, rej) {
request.post({
url: url,
formData: formdata
}, function (error, response, body) {
if (!error && response.statusCode == 200) {
resv(body);
} else {
rej(error)
}
});
})
return promise;
}
/**
* 请求返回成功
* @param {*} data 操作成功返回的数据,有值为成功，无值为失败
* @param {*} okmsg 操作成功的描述
* @param {*} req 请求头信息
*/
static getResult(data, opmsg = "操作成功", req) {
return {
status: !data ? -1 : 1,
message: opmsg,
data: data,
bizmessage: req && req.session && req.session.bizmsg ? req.session.bizmsg : "empty"
};
}
/**
* 请求返回成功
* @param {*} data 操作成功返回的数据
* @param {*} okmsg 操作成功的描述
*/
static getResultSuccess(data, okmsg = "success") {
return {
status: 1,
message: okmsg,
data: data || null,
};
}
/**
* 请求返回失败
* @param {*} status 操作失败状态，默认为-1
* @param {*} errmsg 操作失败的描述，默认为fail
* @param {*} data 操作失败返回的数据
*/
static getResultFail(status = -1, errmsg = "fail", data = null) {
return {
status: status,
message: errmsg,
data: data,
};
}
/**
* 请求处理异常
* @param {*} errmsg 操作失败的描述，默认为fail
* @param {*} data 操作失败返回的数据
*/
static getResultError(errmsg = "fail", data = null) {
return {
status: -200,
message: errmsg,
data: data,
};
}
static register(key, ClassObj, groupName, filename) {
if (System.objTable[key] != null) {
throw new Error("相同key的对象已经存在");
} else {
let obj;
if (ClassObj.name === "ServiceBase") {
obj = new ClassObj(groupName, filename.replace("Sve", "Dao"));
} else {
obj = new ClassObj(groupName, filename);
}
System.objTable[key] = obj;
}
return System.objTable[key];
}
static getObject(objpath) {
var pathArray = objpath.split(".");
var packageName = pathArray[0];
var groupName = pathArray[1];
var filename = pathArray[2];
var classpath = "";
if (filename) {
classpath = objsettings[packageName] + "/" + groupName;
} else {
classpath = objsettings[packageName];
filename = groupName;
}
var objabspath = classpath + "/" + filename + ".js";
//判断文件的存在性
//如果不存在，需要查看packageName
//如果packageName=web.service,dao
if (System.objTable[objabspath] != null) {
return System.objTable[objabspath];
} else {
var ClassObj = null;
try {
ClassObj = require(objabspath);
} catch (e) {
console.log(e.stack,"...getObject.....errror");
let fname = objsettings[packageName + "base"];
ClassObj = require(fname);
}
if (ClassObj.name == "Dao") {
let modelname = filename.substring(0, filename.lastIndexOf("Dao"))
return System.register(objabspath, ClassObj, modelname);
}
if (ClassObj.name.indexOf("Ctl") >= 0) {
console.log(ClassObj.name);
}
return System.register(objabspath, ClassObj, groupName, filename);
}
}
static get_client_ip(req) {
var ip = req.headers['x-forwarded-for'] ||
req.ip ||
req.connection.remoteAddress ||
req.socket.remoteAddress ||
(req.connection.socket && req.connection.socket.remoteAddress) || '';
var x = ip.match(/(25[0-5]|2[0-4][0-9]|[0-1]{1}[0-9]{2}|[1-9]{1}[0-9]{1}|[1-9])\.(25[0-5]|2[0-4][0-9]|[0-1]{1}[0-9]{2}|[1-9]{1}[0-9]{1}|[1-9]|0)\.(25[0-5]|2[0-4][0-9]|[0-1]{1}[0-9]{2}|[1-9]{1}[0-9]{1}|[1-9]|0)\.(25[0-5]|2[0-4][0-9]|[0-1]{1}[0-9]{2}|[1-9]{1}[0-9]{1}|[0-9])$/);
if (x) {
return x[0];
} else {
return "localhost";
}
};
}
Date.prototype.Format = function (fmt) { //author: meizz
var o = {
"M+": this.getMonth() + 1, //月份
"d+": this.getDate(), //日
"h+": this.getHours(), //小时
"m+": this.getMinutes(), //分
"s+": this.getSeconds(), //秒
"q+": Math.floor((this.getMonth() + 3) / 3), //季度
"S": this.getMilliseconds() //毫秒
};
if (/(y+)/.test(fmt))
fmt = fmt.replace(RegExp.$1, (this.getFullYear() + "").substr(4 - RegExp.$1.length));
for (var k in o)
if (new RegExp("(" + k + ")").test(fmt))
fmt = fmt.replace(RegExp.$1, (RegExp.$1.length == 1) ? (o[k]) : (("00" + o[k]).substr(("" + o[k]).length)));
return fmt;
}
System.objTable = {};
module.exports = System;
\ No newline at end of file
queue-center/app/base/utils/esUtils.js deleted 100755 → 0
const system = require("../system");
const util = require('util');
// const axios = require('axios');
class EsUtils {
constructor() {
this.execClient = system.getObject("util.execClient");
this.errorLogDao = system.getObject("db.opLogs.errorLogDao");
this.configInfoDao = system.getObject("db.opLogs.configInfoDao");
}
/**
* 添加es日志
* @param {*} queuedName 队列名称
* @param {*} actionBody 日志信息
*/
async addEsLogs(queuedName, actionBody) {
var configInfoResult = await this.configInfoDao.getList();
if (configInfoResult.status != 1) {
this.errorLogDao.addOpErrorLogs("publicLogsConsumer,configInfo list is empty", actionBody, null, null, 1);
return system.getResultSuccess();
}
var publicLogsEsName = configInfoResult.data.filter(f => f.c_key === "publicLogsEsName");
var publicLogsEsPwd = configInfoResult.data.filter(f => f.c_key === "publicLogsEsPwd");
var publicLogsEsReqUrl = configInfoResult.data.filter(f => f.c_key === "publicLogsEsReqUrl");
if (!publicLogsEsName || !publicLogsEsPwd || !publicLogsEsReqUrl) {
this.errorLogDao.addOpErrorLogs("publicLogsConsumer,es account info is empty", actionBody, null, null, 1);
return system.getResultSuccess();
}
var esIndexName = queuedName + "-" + actionBody.identifyCode || "";
var reqUrl = publicLogsEsReqUrl[0].c_value + esIndexName.toLocaleLowerCase() + "/_doc?pretty";
var params = {
opTitle: actionBody.opTitle || "",
identifyCode: actionBody.identifyCode || "",
messageBody: actionBody.messageBody || "",
resultInfo: actionBody.resultInfo || "",
errorInfo: actionBody.errorInfo || "",
requestId: actionBody.requestId
}
var execResult = await this.execPostEs(queuedName, params, reqUrl, publicLogsEsName[0].c_value, publicLogsEsPwd[0].c_value);
return execResult;
}
/**
* post日志到Es
* @param {*} params 参数
* @param {*} url 地址
*/
async execPostEs(queuedName, params, reqUrl, esName, esPwd) {
try {
// let result = await axios({
// // headers: {'Content-Type': 'application/x-www-form-urlencoded'},
// headers: {
// 'Content-type': 'application/json',
// 'Authorization': 'Basic YWRtaW5lczphZG1pbkdTQmVzLg=='
// },
// method: 'POST',
// url: reqUrl,
// data: JSON.stringify(params),
// timeout: 5000
// });
// if (result.status == 201) {
// return system.getResultSuccess();
// }
// this.errorLogDao.addOpErrorLogs(queuedName + "执行execPostEs存在错误", params, result, null, 3);
// return system.getResult(null, "执行execPostEs存在错误");
//方式二
var result = await this.execClient.execPostEs(params, reqUrl, esName, esPwd);
if (!result || !result.stdout) {
this.errorLogDao.addOpErrorLogs(queuedName + "执行execPostEs存在错误", params, result, null, 3);
return system.getResult(null, "执行execPostEs存在错误");
return system.getResult(null, "execPostTimeOut data is empty");
}
return system.getResultSuccess();
} catch (error) {
console.log(error.stack, "......axios.........");
this.errorLogDao.addOpErrorLogs(queuedName + "执行execPostEs存在异常", params, null, error.stack, 3);
}
}
}
module.exports = EsUtils;
queue-center/app/base/utils/execClient.js deleted 100755 → 0
var childproc = require('child_process');
const util = require('util');
const exec = util.promisify(require('child_process').exec);
class ExecClient {
constructor() {
this.cmdGetPattern = "curl -G -X GET '{url}'";
this.cmdPostPattern = "curl -k -H 'Content-type: application/json' -d '{data}' {url}";
// this.cmdPushDataPostPattern = "curl -k -H 'Content-type: application/json' -H 'token:{tk}' -H 'request-id:{requestId}' -d '{data}' {url}"; //--已废弃
}
/**
* get请求
* @param {*} params 提交的数据-格式JSON
* @param {*} url
* @param {*} headData 请求信息，JSON格式
*/
async execGet(params, url, headData) {
let cmd = this.FetchGetCmd(params, url, headData);
console.log(cmd);
var result = await this.exec(cmd);
return result;
}
/**
*
* @param {*} params 提交的数据JSON格式
* @param {*} url
* @param {*} headData 请求信息，JSON格式
* @param {*} timeOut 超时时间
*/
async execGetTimeOut(params, url, headData, timeOut = 5000) {
//timeOut，单位是毫秒
let cmd = this.FetchGetCmd(params, url, headData);
var options = {
timeout: timeOut,
};
const { stdout, stderr } = await exec(cmd, options);
return { stdout, stderr };
}
/**
* 带超时时间的post请求
* @param {*} params 请求数据-json格式
* @param {*} url 请求地址
* @param {*} ContentType 请求头类型，默认application/json
* @param {*} headData 请求头内容-json格式，如：请求头中传递token,格式：{token:"9098902q849q0434q09439"}
*/
async execPost(params, url, ContentType, headData) {
let cmd = this.FetchPostCmd(params, url, ContentType, headData);
var options = {
maxBuffer: 1024 * 1024 * 15
};
var result = await this.exec(cmd, options, headData);
return result;
}
/**
* 带超时时间的post请求
* @param {*} params 请求数据-json格式
* @param {*} url 请求地址
* @param {*} ContentType 请求头类型，默认application/json
* @param {*} headData 请求头内容-json格式，如：请求头中传递token,格式：{token:"9098902q849q0434q09439"}
* @param {*} timeOut 超时时间设置，单位秒
*/
async execPostTimeOut(params, url, ContentType, headData, timeOut = 5000) {
let cmd = this.FetchPostCmd(params, url, ContentType, headData);
var options = {
timeout: timeOut,
maxBuffer: 1024 * 1024 * 15
};
var result = await this.exec(cmd, options);
return result;
}
/**
* post日志到Es
* @param {*} params 参数
* @param {*} reqUrl
* @param {*} esName
* @param {*} esPwd
*/
async execPostEs(params, reqUrl, esName, esPwd) {
let cmd = this.FetchPostEsCmd(params, reqUrl, esName, esPwd);
var result = await this.exec(cmd);
return result;
}
//--------------------------------------------------辅助方法start-----------------
async exec(cmd) {
//await后面表达式返回的promise对象，是then的语法糖，await返回then函数的返回值
//异常需要try/catch自己捕获或外部catch捕获
const { stdout, stderr } = await exec(cmd);
return { stdout, stderr };
}
FetchGetCmd(params, url, headData) {
var cmd = this.cmdGetPattern.replace(
/\{data\}/g, params).replace(/\{url\}/g, url);
console.log(cmd);
return cmd;
}
FetchPostCmd(params, url, ContentType, headData) {
if (!ContentType) {
ContentType = "application/json";
}
var data = typeof params === 'object' ? JSON.stringify(params) : params;
var cmdStr = "curl -k -H 'Content-type:" + ContentType + "'";
if (headData) {
var headDataKeys = Object.keys(headData);
if (headDataKeys.length > 0) {
for (let index = 0; index < headDataKeys.length; index++) {
const indexKey = headDataKeys[index];
var headValue = headData[indexKey];
if (indexKey && headValue) {
cmdStr = cmdStr + " -H '" + indexKey + ":" + headValue + "'";
}
}
}
}
cmdStr = cmdStr + " -d '" + data + "' " + url;
console.log(cmdStr, "：cmdStr.................");
return cmdStr;
}
FetchPostEsCmd(params, reqUrl, esName, esPwd) {
var data = JSON.stringify(params);
var cmdStr = "curl --user " + esName + ":" + esPwd + " -k -H 'Content-type: application/json' -d '" + data + "' " + reqUrl;
console.log(cmdStr, "：cmdStr.................");
return cmdStr;
}
//--------------------------------------------------辅助方法end-----------------
}
module.exports = ExecClient;
queue-center/app/base/utils/redisClient.js deleted 100755 → 0
const system = require("../system");
const redis = require("redis");
const settings = require("../../config/settings");
const bluebird = require("bluebird");
bluebird.promisifyAll(redis);
class RedisClient {
constructor() {
this.redisConfig = settings.redis();
this.client = this.getCreateClientInstance();
this.subclient = this.client.duplicate();
this.client.on("error", function (err) {
console.log(err, "..redisClient........error");
// //日志记录
// logCtl.error({
// optitle:"redis this.client.on异常：",
// op:"base/utils/redisClient/this.client.on",
// content:err,
// clientIp:""
// });
});
/**
* 管理锁
* @param key 换成key
* @param expires 过期时间
* @returns {Promise<*>}
*/
const setLock = async (key, expires = 5) => {
// 这里锁名修改成taskLocks
const lock = `cron:recyclePool:taskLocks:${key}`;
const res = await this.client.set(lock, key, 'EX', expires, 'NX');
return res;
};
const delLock = async (key) => {
const lock = `cron:recyclePool:taskLocks:${key}`;
await this.client.del(lock);
};
const self = this;
// 监听回调
const subExpired = async (err, res) => {
// 这里需要创建一个新的Redis对象
// 因为 连接在订阅服务器模式下，只能使用订阅服务器命令
const sub = self.getCreateClientInstance();
// 设置事件参数
const expired_subKey = `__keyevent@${self.redisConfig.db}__:expired`
sub.subscribe(expired_subKey, function () {
sub.on('message', async function (channel, msg) {
try {
var opResult = msg.split("$$");
console.log(opResult[0], "..opResult[0]..............");
console.log(opResult[1], "..opResult[1]..............");
if (!opResult || opResult.length == 0) {
return;
}
if (["SYTXFAIL-SYTXPUBLIC-MSGQ"].indexOf(opResult[0]) < 0) {
return;
}
// 锁会在 5 秒后过期
const lock = await setLock(opResult[1]);
if (lock) {
var opData = await self.zrangeList(opResult[0], opResult[1], opResult[1]);
if (!opData || opData.length == 0) {
delLock(opResult[1]);
return;
}
var opDataItem = JSON.parse(opData[0]);
await self.lpushData(opDataItem.queuedName, opDataItem);
await self.zremRangebyscoreData(opResult[0], opResult[1], opResult[1]);
delLock(opResult[1])
}
} catch (error) {
//TODO：日志处理
console.log(error.stack, ".....message......................error...........");
}
});
});
};
// 创建监听
this.client.send_command('config', ['set', 'notify-keyspace-events', 'Ex'], subExpired)
}
getDuplicateInstance() {
return this.subclient;
}
getCreateClientInstance() {
var self = this;
return redis.createClient({
host: self.redisConfig.host,
port: self.redisConfig.port,
password: self.redisConfig.password,
db: self.redisConfig.db,
retry_strategy: function (options) {
if (options.total_retry_time > 1000 * 60 * 60) {
return new Error('Retry time exhausted');
}
if (options.attempt > 10) {
return 10000;
}
return Math.min(options.attempt * 100, 3000);
}
});
}
/**
* 生产供消费者消费的数据
* @param {*} queuedName 队列名称
* @param {*} actionBody 业务数据
*/
lpushData(queuedName, messageBody) {
messageBody = typeof messageBody === 'object' ? JSON.stringify(messageBody) : messageBody;
this.client.lpush(queuedName, messageBody, function (err, reply) {
if (err) {
console.log('lpush message error :' + err + "，queuedName：" + queuedName + "，messageBody：" + messageBody);
return new Error('lpush message error :' + err);
} else {
console.log('lpush message success');
}
});
}
/**
* 设置延迟事件redis存储
* @param {*} key key
* @param {*} exSeconds 过期时间-单位秒
* @param {*} value 值
*/
async setDelayEventData(key, value, exSeconds) {
value = typeof value === 'object' ? JSON.stringify(value) : value;
var key = key + "$$" + value;
this.client.set(key, "delay event", 'EX', exSeconds);
}
/**
* 添加集合数据
* @param {*} collectionName 集合名称
* @param {*} score 分数，用于区间查询
* @param {*} param 参数信息（可以是json）
*/
async zaddSortedSet(collectionName, score, param) {
param = typeof param === 'object' ? JSON.stringify(param) : param;
await this.client.zadd([collectionName, score, param], function (err, data) {
if (err) {
// console.log('zadd data error :' + err);
return new Error('zadd data error :' + err);
} else {
// console.log(data, 'zadd data success');
}
});
}
/**
* 获取集合的区间值
* @param {*} collectionName 集合名称
* @param {*} minValue 最小值
* @param {*} maxValue 最大值
*/
async zrangeList(collectionName, minValue, maxValue) {
return await this.client.zrangebyscoreAsync(collectionName, minValue, maxValue);
}
/**
* 删除集合的区间值
* @param {*} collectionName 集合名称
* @param {*} minValue 最小值
* @param {*} maxValue 最大值
*/
async zremRangebyscoreData(collectionName, minValue, maxValue) {
return await this.client.zremrangebyscoreAsync(collectionName, minValue, maxValue);
}
async setWithEx(key, val, t) {
var p = this.client.setAsync(key, val);
this.client.expire(key, t);
return p;
}
async getCache(key) {
return this.client.getAsync(key);
}
}
module.exports = RedisClient;
queue-center/app/config/environment.js deleted 100755 → 0
var express = require('express');
var path = require('path');
var methodOverride = require('method-override');
// var cookierParser = require('cookie-parser');
// var session = require('express-session');
// var RedisStore = require('connect-redis')(session);
var bodyParser = require('body-parser');
// var multer = require('multer');
var errorHandler = require('errorhandler');
var settings = require('./settings');
var system = require('../base/system');
var routes = require('./routes');
// const logCtl = system.getObject("service.common.oplogSve");
// const clientRedis = system.getObject("util.redisClient").client;
//const tm=system.getObject("db.taskManager");
module.exports = function (app) {
app.set('port', settings.port);
app.use(methodOverride());
// app.use(cookierParser());
app.use(bodyParser.json({ limit: '50mb' }));
app.use(bodyParser.urlencoded({ limit: '50mb', extended: true }));
routes(app);//初始化路由
app.use(express.static(path.join(settings.basepath, '/app/front/entry/public')));
app.all('*', function (req, res, next) {
req.objs = system;
res.header('Access-Control-Allow-Origin', '*');
res.header('Access-Control-Allow-Headers', 'Content-Type, Content-Length, Authorization, Accept, X-Requested-With , yourHeaderFeild');
res.header('Access-Control-Allow-Methods', 'PUT, POST, GET, DELETE, OPTIONS');
// res.header('Access-Control-Allow-Credentials', 'true');
if (req.method == 'OPTIONS') {
res.send(200); //让options请求快速返回/
}
else {
next();
}
});
// if (settings.env == "dev") {
// app.use(session(
// {
// name: 'devdemosid',
// cookie: { maxAge: 3600000 },
// rolling: true,
// resave: false,
// saveUninitialized: false,
// secret: 'uwotm8',
// store: new RedisStore({
// client: clientRedis,
// }),
// }));
// } else {
// app.use(session(
// {
// name: 'demosid',
// cookie: { maxAge: 3600000 },
// rolling: true,
// resave: false,
// saveUninitialized: false,
// secret: 'uwotm8',
// store: new RedisStore({
// client: clientRedis,
// }),
// }));
// }
// development only
if ('development' == app.get('env')) {
app.use(errorHandler());
} else {
app.use(function (err, req, res) {
console.log("prod error handler...........................................");
console.log(err);
logCtl.error({
optitle: "environment 调用异常error:",
op: req.url,
content: e.toString(),
clientIp: system.get_client_ip(req),
agent: req.headers["user-agent"],
});
//logerApp.error("prod error handler",err);
res.send("link index");
});
}
};
queue-center/app/config/objsettings.js deleted 100755 → 0
var path= require('path');
var basepath=path.normalize(path.join(__dirname, '../..'));
var settings = {
api:path.join(basepath,"app/base/api/impl"),
util:path.join(basepath,"app/base/utils"),
service:path.join(basepath,"app/base/service/impl"),
db:path.join(basepath,"app/base/db/impl"),
service2:path.join(basepath,"app/base/service"),
consumer:path.join(basepath,"app/base/db/consumer"),
}
module.exports = settings;
queue-center/app/config/routes.js deleted 100755 → 0
var fs=require("fs");
var path=require("path");
var system = require('../base/system');
module.exports = function (app) {
var routePath=__dirname+"/routes";
fs.readdir(routePath,function(err,rs){
if(rs){
rs.forEach(function(r){
var func=require(routePath+"/"+r);
func.call(null,app);
});
}
});
};
queue-center/app/config/routes/api.js deleted 100755 → 0
var url = require("url");
var system = require("../../base/system");
module.exports = function (app) {
//-----------------------新的模式---------api---------开始
app.all("/api/*", async function (req, res, next) {
var result = system.getResult(null, "req method must is post");
if (req.method != "POST") {
res.end(JSON.stringify(result));
return;
}
// if (["getAppTokenByHosts", "getAppTokenByAppKey"].indexOf(req.body.actionType) >= 0) {
// req.body.actionBody.appHosts = req.host;
// next();
// return;
// }
// if (req.path.indexOf("/taskapi/") >= 0) {
// next();
// return;
// }
if (!req.body.actionType) {
result.msg = "actionType can not be empty";
res.end(JSON.stringify(result));
return;
}
next();
});
app.get('/api/:gname/:qname/:method', function (req, res) {
var classPath = req.params["qname"];
var methodName = req.params["method"];
var gname = req.params["gname"];
classPath = gname + "." + classPath;
var tClientIp = system.get_client_ip(req);
req.clientIp = tClientIp;
req.uagent = req.headers["user-agent"];
req.classname = classPath;
var params = [];
params.push(gname);
params.push(methodName);
params.push(req.body);
params.push(req.query);
params.push(req);
var p = null;
var invokeObj = system.getObject("api." + classPath);
if (invokeObj["doexecMethod"]) {
p = invokeObj["doexecMethod"].apply(invokeObj, params);
}
p.then(r => {
res.end(JSON.stringify(r));
});
});
app.post('/api/:gname/:qname/:method', function (req, res) {
var classPath = req.params["qname"];
var methodName = req.params["method"];
var gname = req.params["gname"];
var params = [];
classPath = gname + "." + classPath;
var tClientIp = system.get_client_ip(req);
req.clientIp = tClientIp;
req.uagent = req.headers["user-agent"];
req.classname = classPath;
params.push(gname);
params.push(methodName);
params.push(req.body);
params.push(req.query);
params.push(req);
var p = null;
var invokeObj = system.getObject("api." + classPath);
if (invokeObj["doexecMethod"]) {
p = invokeObj["doexecMethod"].apply(invokeObj, params);
}
p.then(r => {
res.end(JSON.stringify(r));
});
});
//-----------------------新的模式---------api---------结束
};
queue-center/app/config/settings.js deleted 100755 → 0
var path = require('path');
var ENVINPUT = {
DB_HOST: process.env.DB_HOST,
DB_PORT: process.env.DB_PORT,
DB_USER: process.env.DB_USER,
DB_PWD: process.env.DB_PWD,
REDIS_HOST: process.env.REDIS_HOST,
REDIS_PORT: process.env.REDIS_PORT,
REDIS_PWD: process.env.REDIS_PWD,
REDIS_DB: process.env.TASK_REDIS_DB,
DB_NAME: process.env.TASK_DB_NAME,
APP_ENV: process.env.APP_ENV ? process.env.APP_ENV : "dev",//运行环境
CONSUMER_NAME: process.env.CONSUMER_NAME || "publicServiceAllocation.publicConsumer",//消费者名称
QUEUED_NAME: process.env.QUEUED_NAME || "SYTXPUBLIC-MSGQ",//队列名称，FAIL-失败队列（队列和失败队列一对存在进行部署）
};
var settings = {
env: ENVINPUT.APP_ENV,
consumerName: ENVINPUT.CONSUMER_NAME,
queuedName: ENVINPUT.QUEUED_NAME,
basepath: path.normalize(path.join(__dirname, '../..')),
port: process.env.NODE_PORT || 8080,
redis: function () {
if (this.env == "dev") {
console.log("dev.........................................................");
var localsettings = require("./localsettings");
return localsettings.redis;
} else {
return {
host: ENVINPUT.REDIS_HOST,
port: ENVINPUT.REDIS_PORT,
password: ENVINPUT.REDIS_PWD,
db: ENVINPUT.REDIS_DB,
};
}
},
database: function () {
if (this.env == "dev") {
var localsettings = require("./localsettings");
return localsettings.database;
} else {
return {
dbname: ENVINPUT.DB_NAME,
user: ENVINPUT.DB_USER,
password: ENVINPUT.DB_PWD,
config: {
host: ENVINPUT.DB_HOST,
dialect: 'mysql',
operatorsAliases: false,
pool: {
max: 5,
min: 0,
acquire: 90000000,
idle: 1000000
},
debug: false,
dialectOptions: {
requestTimeout: 999999,
// instanceName:'DEV'
} //设置MSSQL超时时间
},
};
}
}
};
settings.ENVINPUT = ENVINPUT;
module.exports = settings;
queue-center/main.js deleted 100755 → 0
const system = require("./app/base/system");
var settings = require("./app/config/settings");
var http = require('http');
var express = require('express');
var environment = require('./app/config/environment');
const app = express();
// //数据库支持暂缓支持
// var dbf = system.getObject("db.common.connection");
// con = dbf.getCon();
console.log(settings.consumerName, "--consumerName-----start-----");
if (settings.consumerName) {
var consumer = system.getObject("consumer." + settings.consumerName);
(async () => {
await consumer.doConsumer(settings.queuedName);
})();
} else {
console.log("not find consumer,please check ............................");
}
environment(app);//初始化环境
var server = http.createServer(app);
server.listen(settings.port, function () {
console.log('Express server listening on port ' + settings.port);
});
queue-center/package-lock.json deleted 100755 → 0
This source diff could not be displayed because it is too large. You can view the blob instead.
queue-center/package.json deleted 100755 → 0
{
"name": "bigdata",
"version": "1.0.0",
"description": "h5framework",
"main": "main.js",
"scripts": {
"dev": "nodemon main.js",
"test": "echo \"Error: no test specified\" && exit 1"
},
"license": "ISC",
"dependencies": {
"@alicloud/pop-core": "^1.7.7",
"MD5": "^1.3.0",
"after": "^0.8.2",
"ali-oss": "^4.12.2",
"aliyun-api-gateway": "^1.1.6",
"babel-polyfill": "^6.26.0",
"base64id": "^1.0.0",
"bluebird": "^3.5.1",
"body-parser": "^1.18.2",
"co": "^4.6.0",
"connect-redis": "^3.3.3",
"continuation-local-storage": "^3.2.1",
"cookie-parser": "^1.4.3",
"crypto": "^1.0.1",
"crypto-js": "^3.1.9-1",
"ejs": "^2.5.8",
"element-ui": "^2.4.0",
"engine.io-parser": "^2.1.2",
"errorhandler": "^1.5.0",
"exif-js": "^2.3.0",
"express": "^4.16.2",
"express-session": "^1.15.6",
"gm": "^1.23.1",
"marked": "^0.7.0",
"method-override": "^2.3.10",
"morgan": "^1.9.0",
"multer": "^1.3.0",
"mysql2": "^1.5.3",
"node-cron": "^2.0.1",
"node-uuid": "^1.4.8",
"node-xlsx": "^0.15.0",
"nodemailer": "^6.3.0",
"pinyin": "^2.9.0",
"puppeteer": "^1.20.0",
"qr-image": "^3.2.0",
"sequelize": "^4.37.8",
"sequelize-cli": "^4.1.1",
"serve-favicon": "^2.4.5",
"sha1": "^1.1.1",
"socket.io": "^2.1.1",
"uuid": "^3.2.1"
},
"devDependencies": {
"element-theme": "^2.0.1",
"element-theme-chalk": "^2.4.0"
}
}
queue-center/readme.txt deleted 100755 → 0
大家关注：推送时间：0，2，4，6，8
一.公共队列配置
公共业务配置：
消费者名称（CONSUMER_NAME）：publicServiceAllocation.publicConsumer
队列名称 （QUEUED_NAME）：SYTXPUBLIC-MSGQ
公共业务失败配置：
消费者名称（CONSUMER_NAME）：publicFail.publicFailConsumer
队列名称 （QUEUED_NAME）：SYTXFAIL-SYTXPUBLIC-MSGQ
公共日志配置：
消费者名称（CONSUMER_NAME）：publicLogs.publicLogsConsumer
队列名称 （QUEUED_NAME）：LOGS-SYTXPUBLIC-MSGQ
公共通知配置：
通知者名称（CONSUMER_NAME）：publicNotify.publicNotifyConsumer
队列名称 （QUEUED_NAME）：NOTIFY-SYTXPUBLIC-MSGQ
二.公共队列接口
1.生产业务数据供消费者消费
地址：http://192.168.1.189:4016/api/queueAction/producer/springBoard
请求方式：post
参数：
{
"actionType": "produceData",// Y 功能名称
"actionBody": {
"pushUrl": "http://192.168.1.189:4012/api/test/testApi/springBoard",// Y 推送地址
"notifyUrl": "",// N 推送成功后通知的Url
"actionType": "test",// Y 推送地址接收时的功能名称
"identifyCode": "001",// Y 操作的业务标识
"messageBody": { // Y 推送的业务消息，必须有一项对象属性值
"name":"张三",
"age":21
}
}
}
2.生产日志数据供消费者消费
地址：http://192.168.1.189:4017/api/queueAction/producer/springBoard
请求方式：post
参数：
{
"actionType": "produceLogsData",// Y 功能名称
"actionBody": {
"opTitle": "",// N 操作的业务标题
"identifyCode": "logs001",// Y 操作的业务标识
"messageBody": {//日志的描述信息
"opUrl": "http://192.168.1.189:4012/api/test/testApi/springBoard",// N 操作的业务Url
"message": ""// Y 日志消息描述
}
}
}
2.有回调地址的接口通知
地址：http://XXX.com/XXX
请求方式：post
参数：
{
"actionType": "produceLogsData",// Y 功能名称
"identifyCode": "logs001",// Y 操作的业务标识
"requestId":"23ueou23oifdlkfsddslkfja",
"actionBody": {
"messageBody": {
"opUrl": "http://192.168.1.189:4012/api/test/testApi/springBoard",// N 操作的业务Url
"message": ""// Y 日志消息描述
},
"resultInfo":{//返回的结果
}
}
}
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment