MESSAGE_DISPATCH.js 9.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335
  1. const CMD = {}
  2. const time_count = 60000*5 //60秒更新一次配置
  3. const config = require('../etc/config.json')
  4. const redis_help = require('../src/use_redis');
  5. const WebSocketClient = require('../src/WebSocketClient');
  6. const rabbitMq = require('../src/mq/rabbit-mq');
  7. const tools = require('../tools');
  8. console.log("运行 MESSAGE_DISPATCH")
  9. CMD.updateFilterConfig = async function(){
  10. const postData = {
  11. cmd:"db_filter_config",
  12. fun:"getConfig",
  13. data:{}
  14. };
  15. let client = tools.getOneNewClinet()
  16. client.post('http://127.0.0.1:9100/tg/back/api', postData)
  17. .then(response => {
  18. redis_help.setKeyValue("FilterConfig",JSON.stringify(response.data.data))
  19. })
  20. .catch(error => {
  21. console.error('updateFilterConfig error!', error);
  22. });
  23. }
  24. CMD.updateCloseMainConfig = async function(){
  25. const postData = {
  26. cmd:"tg_main",
  27. fun:"getCloseCreateLinkMainList",
  28. data:{}
  29. };
  30. let client = tools.getOneNewClinet()
  31. client.post('http://127.0.0.1:9100/tg/back/api', postData)
  32. .then(response => {
  33. redis_help.setKeyValue("CloseMainConfig",JSON.stringify(response.data.data))
  34. })
  35. .catch(error => {
  36. console.error('updateCloseMainConfig error!', error);
  37. });
  38. }
  39. CMD.updateMainConfig = async function(){
  40. const postData = {
  41. cmd:"tg_main",
  42. fun:"getOpenCreateLinkMainList",
  43. data:{}
  44. };
  45. let client = tools.getOneNewClinet()
  46. client.post('http://127.0.0.1:9100/tg/back/api', postData)
  47. .then(response => {
  48. CMD.updateCloseMainConfig()
  49. redis_help.setKeyValue("MainConfig",JSON.stringify(response.data.data))
  50. })
  51. .catch(error => {
  52. console.error('updateFilterConfig error!', error);
  53. });
  54. }
  55. CMD.updateAppConfig = async function(){
  56. const postData = {
  57. cmd:"tg_app",
  58. fun:"getAppList",
  59. data:{}
  60. };
  61. let client = tools.getOneNewClinet()
  62. client.post('http://127.0.0.1:9100/tg/back/api', postData)
  63. .then(response => {
  64. redis_help.setKeyValue("AppConfig",JSON.stringify(response.data.data))
  65. })
  66. .catch(error => {
  67. console.error('updateFilterConfig error!', error);
  68. });
  69. }
  70. CMD.updatePlatformConfig = async function(){
  71. const postData = {
  72. cmd:"tg_platform",
  73. fun:"getPlatformList",
  74. data:{}
  75. };
  76. let client = tools.getOneNewClinet()
  77. client.post('http://127.0.0.1:9100/tg/back/api', postData)
  78. .then(response => {
  79. CMD.updatePlatformShared()
  80. redis_help.setKeyValue("PlatformConfig",JSON.stringify(response.data.data))
  81. })
  82. .catch(error => {
  83. console.error('updateFilterConfig error!', error);
  84. });
  85. }
  86. CMD.updatePlatformShared = async function(){
  87. const postData = {
  88. cmd:"platform_shared",
  89. fun:"gePlatformSharedList",
  90. data:{}
  91. };
  92. let client = tools.getOneNewClinet()
  93. client.post('http://127.0.0.1:9100/tg/back/api', postData)
  94. .then(response => {
  95. redis_help.setKeyValue("PlatformShared",JSON.stringify(response.data.data))
  96. })
  97. .catch(error => {
  98. console.error('PlatformShared error!', error);
  99. });
  100. }
  101. CMD.updateFqMfKeyList = async function(){
  102. const postData = {
  103. cmd:"fq_mf_book",
  104. fun:"get_all_fq_mf_key",
  105. data:{}
  106. };
  107. let client = tools.getOneNewClinet()
  108. client.post('http://127.0.0.1:9100/tg/back/api', postData)
  109. .then(response => {
  110. redis_help.setKeyValue("all_fq_mf_key",JSON.stringify(response.data.data))
  111. })
  112. .catch(error => {
  113. console.error('all_fq_mf_key error!', error);
  114. });
  115. }
  116. CMD.updateFqKeyList = async function(){
  117. const postData = {
  118. cmd:"fq_book",
  119. fun:"get_all_fq_key",
  120. data:{}
  121. };
  122. let client = tools.getOneNewClinet()
  123. client.post('http://127.0.0.1:9100/tg/back/api', postData)
  124. .then(response => {
  125. redis_help.setKeyValue("all_fq_key",JSON.stringify(response.data.data))
  126. })
  127. .catch(error => {
  128. console.error('all_fq_key error!', error);
  129. });
  130. }
  131. CMD.updateBlackBooks = async function(){
  132. const postData = {
  133. cmd:"book_black_list",
  134. fun:"get_all_black_books",
  135. data:{}
  136. };
  137. let client = tools.getOneNewClinet()
  138. client.post('http://127.0.0.1:9100/tg/back/api', postData)
  139. .then(response => {
  140. redis_help.setKeyValue("all_black_books",JSON.stringify(response.data.data))
  141. })
  142. .catch(error => {
  143. console.error('all_black_books error!', error);
  144. });
  145. }
  146. CMD.getPullDataConfig = async function() {
  147. const postData = {
  148. cmd:"pull_data_config",
  149. fun:"get_all_pull_data_list",
  150. data:{}
  151. };
  152. let client = tools.getOneNewClinet()
  153. return await client.post('http://127.0.0.1:9100/tg/back/api', postData)
  154. .then(response => {
  155. redis_help.setKeyValue("PullDataConfig",JSON.stringify(response.data.data))
  156. })
  157. .catch(error => {
  158. console.error('getPullDataConfig error!', error);
  159. });
  160. }
  161. var ws_isOpen = false
  162. CMD.myListener = {
  163. onOpen: function (webSocket, response) {
  164. ws_isOpen = true
  165. //打开链接后,想服务器端发送一条消息
  166. sendMessage("status_task",{work:1})
  167. },
  168. onMessage: function (webSocket, msg) { //msg可能是字符串,也可能是byte数组,取决于服务器送的内容
  169. recvMessage(msg)
  170. },
  171. onClosing: function (webSocket, code, response) {
  172. ws_isOpen = false
  173. },
  174. onClosed: function (webSocket, code, response) {
  175. ws_isOpen = false
  176. },
  177. onFailure: function (webSocket, t, response) {
  178. }
  179. }
  180. function sendMessage(cmd,data){
  181. if(ws_isOpen==true){
  182. var json = {};
  183. json.cmd=cmd; //方法
  184. json.data= data; //参数
  185. var msg=JSON.stringify(json);
  186. if(CMD.wsClient!=undefined&&CMD.wsClient!=null){
  187. CMD.wsClient.sendMsg(msg);
  188. }
  189. }
  190. }
  191. async function sendMQMessage(message,routingKey = null) {
  192. try {
  193. if(routingKey!=null){
  194. await rabbitMq.producerDirectMsg( message,"exchange_system",routingKey);
  195. }else{
  196. await rabbitMq.producerDirectMsg( message,"exchange_system");
  197. }
  198. console.log('消息发送成功');
  199. } catch (error) {
  200. console.error('发送消息失败:', error);
  201. }
  202. }
  203. async function sendPullMQMessage(message,routingKey = null) {
  204. try {
  205. if(routingKey!=null){
  206. await rabbitMq.producerDirectMsg( message,"exchange_update_pull_config",routingKey);
  207. }else{
  208. await rabbitMq.producerDirectMsg( message,"exchange_update_pull_config");
  209. }
  210. console.log('消息发送成功');
  211. } catch (error) {
  212. console.error('发送消息失败:', error);
  213. }
  214. }
  215. async function recvMessage(data){
  216. try {
  217. // 单个任务
  218. let json_msg = JSON.parse(data)
  219. console.log("recvMessage:",json_msg.cmd)
  220. switch (json_msg.cmd) {
  221. case "updateFilterConfig":
  222. CMD.updateFilterConfig()
  223. break;
  224. case "updateMainConfig":
  225. CMD.updateMainConfig()
  226. break;
  227. case "updateAppConfig":
  228. CMD.updateAppConfig()
  229. break;
  230. case "updatePlatformConfig":
  231. CMD.updatePlatformConfig()
  232. break;
  233. case "updateFqKeyList":
  234. CMD.updateFqKeyList()
  235. break;
  236. case "updateFqMfKeyList":
  237. CMD.updateFqMfKeyList()
  238. break;
  239. case "updateBlackBooks":
  240. CMD.updateBlackBooks()
  241. break;
  242. // case "addMain":
  243. case "syncMain":
  244. {
  245. const postData = {
  246. cmd:"filter_task",
  247. fun:json_msg.cmd,
  248. data:json_msg
  249. };
  250. sendMQMessage(postData,json_msg.cmd)
  251. }
  252. break;
  253. case "getPullDataConfig": //获取拉取数据类配置
  254. {
  255. CMD.getPullDataConfig()
  256. }
  257. break;
  258. case "updatePullConig": //同步拉取配置
  259. {
  260. await CMD.getPullDataConfig()
  261. await new Promise(resolve=>setTimeout(resolve,100))
  262. const postData = {
  263. cmd:"filter_task",
  264. fun:json_msg.cmd,
  265. data:json_msg
  266. };
  267. sendPullMQMessage(postData,json_msg.cmd)
  268. }
  269. break;
  270. default:
  271. // const client = new HttpClient({
  272. // timeout: 5000,
  273. // // headers: {
  274. // // 'User-Agent': 'Custom User Agent'
  275. // // }
  276. // });
  277. // const response = await client.post('http://127.0.0.1:9111',postData);
  278. break;
  279. }
  280. }catch(e){
  281. console.error("recvMessage error1!:",e)
  282. }
  283. }
  284. CMD.init = function(){
  285. redis_help.connect((results)=>{
  286. if(results){
  287. CMD.wsClient = new WebSocketClient(config.ws_config.host, {
  288. reconnectDelay: 5000, // 5 seconds
  289. heartbeatInterval: 30000, // 30 seconds
  290. heartbeatTimeout: 5000 // 5 seconds
  291. },CMD.myListener);
  292. }
  293. })
  294. CMD.updateConfig()
  295. CMD.updateTimeConfig()
  296. }
  297. CMD.updateTimeConfig = function(){
  298. let create_heiyan_book_link = require('../src/api/hy/create_heiyan_book_link')
  299. create_heiyan_book_link.login((data)=>{
  300. console.log("hei_yan_token:",data['token'])
  301. redis_help.setKeyValue("hei_yan_token",data['token'])
  302. })
  303. let qm_login = require('../src/api/qm/qm_login')
  304. qm_login.login((data,headers)=>{
  305. console.log("qi_mao_token:",headers['n-token'])
  306. redis_help.setKeyValue("qi_mao_token",headers['n-token'])
  307. })
  308. }
  309. //定时更新配置
  310. CMD.updateConfig = function(){
  311. setInterval(()=>{
  312. CMD.updateTimeConfig()
  313. },time_count)
  314. }
  315. CMD.init()