rabbitmq.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660
  1. package initialize
  2. import (
  3. "encoding/base64"
  4. "fmt"
  5. "mtp20access/global"
  6. rsp "mtp20access/model/mq/response"
  7. "mtp20access/packet"
  8. "mtp20access/res/pb"
  9. "mtp20access/utils"
  10. "github.com/golang/protobuf/proto"
  11. "github.com/streadway/amqp"
  12. "go.uber.org/zap"
  13. "google.golang.org/protobuf/encoding/protojson"
  14. )
  15. func RabbitMQ() *global.RabbitMQ {
  16. url := global.M2A_CONFIG.Rabbitmq.Url
  17. connection, err := amqp.Dial(url)
  18. if err != nil {
  19. global.M2A_LOG.Error("rabbitmq connect failed, err:", zap.Error(err))
  20. return nil
  21. }
  22. channel, err := connection.Channel()
  23. if err != nil {
  24. global.M2A_LOG.Error("rabbitmq open channel failed, err:", zap.Error(err))
  25. return nil
  26. }
  27. global.M2A_LOG.Info("rabbitmq connect successed.")
  28. return &global.RabbitMQ{
  29. Connection: connection,
  30. Channel: channel,
  31. }
  32. }
  33. // MQProc 消息处理对象
  34. type MQProc struct{}
  35. // process 消息处理接口
  36. func (t *MQProc) Process(topic, queuename string, msg *[]byte) {
  37. // info := fmt.Sprintf("rabbitmq receive message from: topic[%s] queue[%s] contentLen[%d]",
  38. // topic,
  39. // queuename,
  40. // len(string(*msg)))
  41. // global.M2A_LOG.Info(info)
  42. if funcode, sessionId, bytes, serialNumber, err := t.getRspProtobuf(msg); err == nil && bytes != nil {
  43. // 尝试获取对应异步任务
  44. if client, exists := global.M2A_Clients[int(sessionId)]; exists {
  45. key := fmt.Sprintf("%v_%v_%v", sessionId, funcode, serialNumber)
  46. // 银行服务相关的回复流水号是错误的,所以需要特殊处理
  47. if int(funcode) == global.T2bBankSignRsp ||
  48. int(funcode) == global.T2bBankCancelSignRsp ||
  49. int(funcode) == global.T2bBankWithdrawRsp ||
  50. int(funcode) == global.T2bBankDepositRsp {
  51. key = fmt.Sprintf("%v_%v", sessionId, funcode)
  52. }
  53. asyncTask := client.GetAsyncTask(key)
  54. if asyncTask != nil {
  55. rspData := string(*bytes)
  56. // 判断是否要加密
  57. if asyncTask.IsEncrypted {
  58. if b, err := packet.Encrypt(*bytes, packet.AESKey, true); err != nil {
  59. global.M2A_LOG.Error("总线回复数据加密失败", zap.Error(err))
  60. return
  61. } else {
  62. rspData = base64.StdEncoding.EncodeToString(b)
  63. }
  64. }
  65. // 给客户端回调
  66. global.M2A_LOG.Info("[S->C]", zap.Any("rsp", funcode), zap.Any("sessionId", sessionId), zap.Any("data", string(rspData)))
  67. r := rsp.MQBodyRsp{
  68. FunCode: funcode,
  69. IsEncrypted: asyncTask.IsEncrypted,
  70. Data: rspData,
  71. }
  72. asyncTask.Rsp <- r
  73. } else {
  74. global.M2A_LOG.Info("找不到对应KEY的异步任务", zap.Any("key", key), zap.Any("AsyncMap", client.GetAllAsyncTask()))
  75. }
  76. } else {
  77. global.M2A_LOG.Info("找不到对应的client", zap.Any("sessionId", sessionId))
  78. }
  79. }
  80. }
  81. // getRspProtobuf 将总线回复的数据反序列化为Protobuf
  82. func (t *MQProc) getRspProtobuf(msg *[]byte) (funcode uint32, sessionId uint32, bytes *[]byte, serialNumber uint32, err error) {
  83. // 分解总线包信息
  84. funcode = utils.BytesToUint32((*msg)[0:4])
  85. sessionId = utils.BytesToUint32((*msg)[4:8])
  86. b := (*msg)[8:]
  87. global.M2A_LOG.Info("收到总线订阅消息", zap.Any("funcode", funcode), zap.Any("sessionId", sessionId), zap.Any("len", len(b)))
  88. switch int(funcode) {
  89. case global.ModifyPwdRsp: // 修改账户密码应答
  90. var p pb.ModifyPwdRsp
  91. if err = proto.Unmarshal(b, &p); err != nil {
  92. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  93. return
  94. }
  95. if bs, e := protojson.Marshal(&p); e != nil {
  96. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  97. return
  98. } else {
  99. bytes = &bs
  100. serialNumber = p.GetHeader().GetRequestID()
  101. }
  102. case global.UserReceiveInfoRsp: // 新增修改收货地址请求响应
  103. var p pb.UserReceiveInfoRsp
  104. if err = proto.Unmarshal(b, &p); err != nil {
  105. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  106. return
  107. }
  108. if bs, e := protojson.Marshal(&p); e != nil {
  109. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  110. return
  111. } else {
  112. bytes = &bs
  113. serialNumber = p.GetHeader().GetRequestID()
  114. }
  115. case global.DelUserReceiveInfoRsp: // 删除收货地址请求响应
  116. var p pb.DelUserReceiveInfoRsp
  117. if err = proto.Unmarshal(b, &p); err != nil {
  118. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  119. return
  120. }
  121. if bs, e := protojson.Marshal(&p); e != nil {
  122. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  123. return
  124. } else {
  125. bytes = &bs
  126. serialNumber = p.GetHeader().GetRequestID()
  127. }
  128. case global.UserReceiptInfoRsp: // 新增修改用户发票信息请求响应
  129. var p pb.UserReceiptInfoRsp
  130. if err = proto.Unmarshal(b, &p); err != nil {
  131. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  132. return
  133. }
  134. if bs, e := protojson.Marshal(&p); e != nil {
  135. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  136. return
  137. } else {
  138. bytes = &bs
  139. serialNumber = p.GetHeader().GetRequestID()
  140. }
  141. case global.DelUserReceiptInfoRsp: // 删除用户发票信息请求响应
  142. var p pb.DelUserReceiptInfoRsp
  143. if err = proto.Unmarshal(b, &p); err != nil {
  144. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  145. return
  146. }
  147. if bs, e := protojson.Marshal(&p); e != nil {
  148. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  149. return
  150. } else {
  151. bytes = &bs
  152. serialNumber = p.GetHeader().GetRequestID()
  153. }
  154. case global.T2bBankSignRsp: // 签约应答
  155. var p pb.T2BBankSignRsp
  156. if err = proto.Unmarshal(b, &p); err != nil {
  157. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  158. return
  159. }
  160. if bs, e := protojson.Marshal(&p); e != nil {
  161. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  162. return
  163. } else {
  164. bytes = &bs
  165. serialNumber = p.GetHeader().GetRequestID()
  166. }
  167. case global.T2bBankCancelSignRsp: // 解约应答
  168. var p pb.T2BBankCancelSignRsp
  169. if err = proto.Unmarshal(b, &p); err != nil {
  170. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  171. return
  172. }
  173. if bs, e := protojson.Marshal(&p); e != nil {
  174. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  175. return
  176. } else {
  177. bytes = &bs
  178. serialNumber = p.GetHeader().GetRequestID()
  179. }
  180. case global.T2bBankWithdrawRsp: // 出金应答
  181. var p pb.T2BBankWithdrawRsp
  182. if err = proto.Unmarshal(b, &p); err != nil {
  183. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  184. return
  185. }
  186. if bs, e := protojson.Marshal(&p); e != nil {
  187. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  188. return
  189. } else {
  190. bytes = &bs
  191. serialNumber = p.GetHeader().GetRequestID()
  192. }
  193. case global.T2bBankDepositRsp: // 入金应答
  194. var p pb.T2BBankDepositRsp
  195. if err = proto.Unmarshal(b, &p); err != nil {
  196. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  197. return
  198. }
  199. if bs, e := protojson.Marshal(&p); e != nil {
  200. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  201. return
  202. } else {
  203. bytes = &bs
  204. serialNumber = p.GetHeader().GetRequestID()
  205. }
  206. case global.WarehouseApplyRsp: // 仓库申请应答
  207. var p pb.WarehouseApplyRsp
  208. if err = proto.Unmarshal(b, &p); err != nil {
  209. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  210. return
  211. }
  212. if bs, e := protojson.Marshal(&p); e != nil {
  213. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  214. return
  215. } else {
  216. bytes = &bs
  217. serialNumber = p.GetHeader().GetRequestID()
  218. }
  219. case global.PerformanceContractedApplyRsp: // 违约申请应答
  220. var p pb.PerformanceContractedApplyRsp
  221. if err = proto.Unmarshal(b, &p); err != nil {
  222. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  223. return
  224. }
  225. if bs, e := protojson.Marshal(&p); e != nil {
  226. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  227. return
  228. } else {
  229. bytes = &bs
  230. serialNumber = p.GetHeader().GetRequestID()
  231. }
  232. case global.PerformanceDelayApplyRsp: // 延期申请应答
  233. var p pb.PerformanceDelayApplyRsp
  234. if err = proto.Unmarshal(b, &p); err != nil {
  235. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  236. return
  237. }
  238. if bs, e := protojson.Marshal(&p); e != nil {
  239. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  240. return
  241. } else {
  242. bytes = &bs
  243. serialNumber = p.GetHeader().GetRequestID()
  244. }
  245. case global.PerformanceManualConfirmRsp: // 履约手动确认应答
  246. var p pb.PerformanceManualConfirmRsp
  247. if err = proto.Unmarshal(b, &p); err != nil {
  248. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  249. return
  250. }
  251. if bs, e := protojson.Marshal(&p); e != nil {
  252. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  253. return
  254. } else {
  255. bytes = &bs
  256. serialNumber = p.GetHeader().GetRequestID()
  257. }
  258. case global.PerformanceModifyContactRsp: // 履约修改联络信息回应
  259. var p pb.PerformanceModifyContactRsp
  260. if err = proto.Unmarshal(b, &p); err != nil {
  261. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  262. return
  263. }
  264. if bs, e := protojson.Marshal(&p); e != nil {
  265. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  266. return
  267. } else {
  268. bytes = &bs
  269. serialNumber = p.GetHeader().GetRequestID()
  270. }
  271. case global.GoodsFavoriteOperateRsp: // 商品收藏操作接口应答
  272. var p pb.GoodsFavoriteOperateRsp
  273. if err = proto.Unmarshal(b, &p); err != nil {
  274. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  275. return
  276. }
  277. if bs, e := protojson.Marshal(&p); e != nil {
  278. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  279. return
  280. } else {
  281. bytes = &bs
  282. serialNumber = p.GetHeader().GetRequestID()
  283. }
  284. case global.ZSBuyOrderListingRsp: // 钻石买挂牌接口应答
  285. var p pb.ZSBuyOrderListingRsp
  286. if err = proto.Unmarshal(b, &p); err != nil {
  287. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  288. return
  289. }
  290. if bs, e := protojson.Marshal(&p); e != nil {
  291. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  292. return
  293. } else {
  294. bytes = &bs
  295. serialNumber = p.GetHeader().GetRequestID()
  296. }
  297. case global.ZSSellOrderListingRsp: // 钻石卖挂牌接口响应
  298. var p pb.ZSSellOrderListingRsp
  299. if err = proto.Unmarshal(b, &p); err != nil {
  300. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  301. return
  302. }
  303. if bs, e := protojson.Marshal(&p); e != nil {
  304. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  305. return
  306. } else {
  307. bytes = &bs
  308. serialNumber = p.GetHeader().GetRequestID()
  309. }
  310. case global.ZSBuyOrderDestingRsp: // 钻石卖摘牌申请接口响应
  311. var p pb.ZSBuyOrderDestingRsp
  312. if err = proto.Unmarshal(b, &p); err != nil {
  313. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  314. return
  315. }
  316. if bs, e := protojson.Marshal(&p); e != nil {
  317. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  318. return
  319. } else {
  320. bytes = &bs
  321. serialNumber = p.GetHeader().GetRequestID()
  322. }
  323. case global.ZSSellOrderDestingApplyOperateRsp: // 钻石卖摘牌申请操作接口应答
  324. var p pb.ZSSellOrderDestingApplyOperateRsp
  325. if err = proto.Unmarshal(b, &p); err != nil {
  326. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  327. return
  328. }
  329. if bs, e := protojson.Marshal(&p); e != nil {
  330. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  331. return
  332. } else {
  333. bytes = &bs
  334. serialNumber = p.GetHeader().GetRequestID()
  335. }
  336. case global.ZSBuyOrderDestingNegPriceRsp: // 买摘牌询价接口应答
  337. var p pb.ZSBuyOrderDestingNegPriceRsp
  338. if err = proto.Unmarshal(b, &p); err != nil {
  339. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  340. return
  341. }
  342. if bs, e := protojson.Marshal(&p); e != nil {
  343. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  344. return
  345. } else {
  346. bytes = &bs
  347. serialNumber = p.GetHeader().GetRequestID()
  348. }
  349. case global.ZSBuyOrderDestingNegPriceOperateRsp: // 买摘牌询价操作接口应答
  350. var p pb.ZSBuyOrderDestingNegPriceOperateRsp
  351. if err = proto.Unmarshal(b, &p); err != nil {
  352. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  353. return
  354. }
  355. if bs, e := protojson.Marshal(&p); e != nil {
  356. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  357. return
  358. } else {
  359. bytes = &bs
  360. serialNumber = p.GetHeader().GetRequestID()
  361. }
  362. case global.WRListingCancelOrderRsp: // 挂牌撤单应答
  363. var p pb.WRListingCancelOrderRsp
  364. if err = proto.Unmarshal(b, &p); err != nil {
  365. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  366. return
  367. }
  368. if bs, e := protojson.Marshal(&p); e != nil {
  369. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  370. return
  371. } else {
  372. bytes = &bs
  373. serialNumber = p.GetHeader().GetRequestID()
  374. }
  375. case global.ReceiptZSOutApplyRsp: // 钻石出库申请接口响应
  376. var p pb.ReceiptZSOutApplyRsp
  377. if err = proto.Unmarshal(b, &p); err != nil {
  378. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  379. return
  380. }
  381. if bs, e := protojson.Marshal(&p); e != nil {
  382. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  383. return
  384. } else {
  385. bytes = &bs
  386. serialNumber = p.GetHeader().GetRequestID()
  387. }
  388. case global.SpotPresaleListingOrderRsp: // 铁合金现货预售挂牌接口应答
  389. var p pb.SpotPresaleListingOrderRsp
  390. if err = proto.Unmarshal(b, &p); err != nil {
  391. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  392. return
  393. }
  394. if bs, e := protojson.Marshal(&p); e != nil {
  395. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  396. return
  397. } else {
  398. bytes = &bs
  399. serialNumber = p.GetHeader().GetRequestID()
  400. }
  401. case global.CJJCMemberOperateRsp: // 出境检测会员操作响应
  402. var p pb.CJJCMemberOperateRsp
  403. if err = proto.Unmarshal(b, &p); err != nil {
  404. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  405. return
  406. }
  407. if bs, e := protojson.Marshal(&p); e != nil {
  408. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  409. return
  410. } else {
  411. bytes = &bs
  412. serialNumber = p.GetHeader().GetRequestID()
  413. }
  414. case global.BSFWMemberOperateRsp: // 保税服务会员操作响应
  415. var p pb.BSFWMemberOperateRsp
  416. if err = proto.Unmarshal(b, &p); err != nil {
  417. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  418. return
  419. }
  420. if bs, e := protojson.Marshal(&p); e != nil {
  421. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  422. return
  423. } else {
  424. bytes = &bs
  425. serialNumber = p.GetHeader().GetRequestID()
  426. }
  427. case global.AccountFundInfoRsp: // 账户资金信息响应
  428. var p pb.AccountFundInfoRsp
  429. if err = proto.Unmarshal(b, &p); err != nil {
  430. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  431. return
  432. }
  433. if bs, e := protojson.Marshal(&p); e != nil {
  434. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  435. return
  436. } else {
  437. bytes = &bs
  438. serialNumber = p.GetHeader().GetRequestID()
  439. }
  440. case global.GZPresaleApplyRsp: // 广钻预售申请响应
  441. var p pb.GZPresaleApplyRsp
  442. if err = proto.Unmarshal(b, &p); err != nil {
  443. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  444. return
  445. }
  446. if bs, e := protojson.Marshal(&p); e != nil {
  447. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  448. return
  449. } else {
  450. bytes = &bs
  451. serialNumber = p.GetHeader().GetRequestID()
  452. }
  453. case global.GZPresaleOrderRsp: // 广钻预售认购下单响应
  454. var p pb.GZPresaleOrderRsp
  455. if err = proto.Unmarshal(b, &p); err != nil {
  456. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  457. return
  458. }
  459. if bs, e := protojson.Marshal(&p); e != nil {
  460. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  461. return
  462. } else {
  463. bytes = &bs
  464. serialNumber = p.GetHeader().GetRequestID()
  465. }
  466. case global.GZCenterPurchaseApplyRsp: // 广钻集采申请响应
  467. var p pb.GZCenterPurchaseApplyRsp
  468. if err = proto.Unmarshal(b, &p); err != nil {
  469. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  470. return
  471. }
  472. if bs, e := protojson.Marshal(&p); e != nil {
  473. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  474. return
  475. } else {
  476. bytes = &bs
  477. serialNumber = p.GetHeader().GetRequestID()
  478. }
  479. case global.GZCenterPurchaseOrderRsp: // 广钻集采认购下单响应
  480. var p pb.GZCenterPurchaseOrderRsp
  481. if err = proto.Unmarshal(b, &p); err != nil {
  482. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  483. return
  484. }
  485. if bs, e := protojson.Marshal(&p); e != nil {
  486. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  487. return
  488. } else {
  489. bytes = &bs
  490. serialNumber = p.GetHeader().GetRequestID()
  491. }
  492. case global.BSWMSReckonPayRsp: // WMS结算单支付接口响应
  493. var p pb.BSWMSReckonPayRsp
  494. if err = proto.Unmarshal(b, &p); err != nil {
  495. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  496. return
  497. }
  498. if bs, e := protojson.Marshal(&p); e != nil {
  499. global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
  500. return
  501. } else {
  502. bytes = &bs
  503. serialNumber = p.GetHeader().GetRequestID()
  504. }
  505. }
  506. return
  507. }
  508. // RabbitMQSubscribeTopic 订阅主题
  509. func RabbitMQSubscribeTopic() (err error) {
  510. // 订阅需要的总线响应主题
  511. if err = global.SubscribeTopic(global.TOPIC_RSP_QKERNEL); err != nil {
  512. global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
  513. return
  514. }
  515. if err = global.SubscribeTopic(global.TOPIC_MANAGE_RSP); err != nil {
  516. global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
  517. return
  518. }
  519. if err = global.SubscribeTopic(global.TOPIC_RSP_BANK); err != nil {
  520. global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
  521. return
  522. }
  523. if err = global.SubscribeTopic(global.TOPIC_RSP_MONEY); err != nil {
  524. global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
  525. return
  526. }
  527. if err = global.SubscribeTopic(global.TOPIC_PERFORMANCE_RSP); err != nil {
  528. global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
  529. return
  530. }
  531. if err = global.SubscribeTopic(global.TOPIC_WAREHOUSE_RECIEPT_RSP); err != nil {
  532. global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
  533. return
  534. }
  535. if err = global.SubscribeTopic(global.TOPIC_RSP_WAREHOUSE_TRADE); err != nil {
  536. global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
  537. return
  538. }
  539. if err = global.SubscribeTopic(global.TOPIC_RSP_WAREHOUSE_TRADE_GZ); err != nil {
  540. global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
  541. return
  542. }
  543. if err = global.SubscribeTopic(global.TOPIC_RSP_CJBS_TRADE_GZ); err != nil {
  544. global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
  545. return
  546. }
  547. // 铁合金
  548. if err = global.SubscribeTopic(global.TOPIC_RSP_WAREHOUSE_PRESALE_THJ); err != nil {
  549. global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
  550. return
  551. }
  552. if err = global.SubscribeTopic(global.TOPIC_MANAGER_THJ_NTF); err != nil {
  553. global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
  554. return
  555. }
  556. if err = global.SubscribeTopic(global.TOPIC_RSP_WAREHOUSE_PRESALE_GZ); err != nil {
  557. global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
  558. return
  559. }
  560. if err = global.SubscribeTopic(global.TOPIC_RSP_WAREHOUSE_CENTERPURCHASE_GZ); err != nil {
  561. global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
  562. return
  563. }
  564. global.M2A_LOG.Info("rabbitmq subscribe topic successed.")
  565. return
  566. }
  567. // StartRabbitMQReceive 开始接收总线消息
  568. func StartRabbitMQReceive() {
  569. t := &MQProc{}
  570. go func() {
  571. for _, subinfo := range global.SubInfos {
  572. global.Receive(subinfo.Topic, subinfo.QueueName, t)
  573. }
  574. }()
  575. }
  576. // InitFuncodeTopic 初始化功能码主题MAP
  577. func InitFuncodeTopic() {
  578. if global.M2A_FuncodeTopic == nil {
  579. global.M2A_FuncodeTopic = make(map[string][]int)
  580. }
  581. global.M2A_FuncodeTopic[global.TOPIC_REQ_QKERNEL] = []int{
  582. global.ModifyPwdReq,
  583. }
  584. global.M2A_FuncodeTopic[global.TOPIC_MANAGE_REQ] = []int{
  585. global.UserReceiveInfoReq,
  586. global.DelUserReceiveInfoReq,
  587. global.UserReceiveIsDefaultReq,
  588. global.UserReceiptInfoReq,
  589. global.DelUserReceiptInfoReq,
  590. global.WarehouseApplyReq,
  591. }
  592. global.M2A_FuncodeTopic[global.TOPIC_REQ_BANK] = []int{
  593. global.T2bBankSignReq,
  594. global.T2bBankCancelSignReq,
  595. global.T2bBankWithdrawReq,
  596. global.T2bBankDepositReq,
  597. }
  598. global.M2A_FuncodeTopic[global.TOPIC_REQ_MONEY] = []int{
  599. global.AccountFundInfoReq,
  600. }
  601. global.M2A_FuncodeTopic[global.TOPIC_PERFORMANCE_REQ] = []int{
  602. global.PerformanceContractedApplyReq,
  603. global.PerformanceDelayApplyReq,
  604. global.PerformanceManualConfirmReq,
  605. global.PerformanceModifyContactReq,
  606. }
  607. global.M2A_FuncodeTopic[global.TOPIC_WAREHOUSE_RECIEPT] = []int{
  608. global.ReceiptZSOutApplyReq,
  609. }
  610. global.M2A_FuncodeTopic[global.TOPIC_REQ_WAREHOUSE_TRADE] = []int{
  611. global.WRListingCancelOrderReq,
  612. }
  613. global.M2A_FuncodeTopic[global.TOPIC_REQ_WAREHOUSE_TRADE_GZ] = []int{
  614. global.GoodsFavoriteOperateReq,
  615. global.ZSBuyOrderListingReq,
  616. global.ZSSellOrderListingReq,
  617. global.ZSBuyOrderDestingReq,
  618. global.ZSSellOrderDestingApplyReq,
  619. global.ZSSellOrderDestingApplyOperateReq,
  620. global.ZSBuyOrderDestingNegPriceReq,
  621. global.ZSBuyOrderDestingNegPriceOperateReq,
  622. }
  623. global.M2A_FuncodeTopic[global.TOPIC_REQ_CJBS_TRADE_GZ] = []int{
  624. global.CJJCMemberOperateReq,
  625. global.BSFWMemberOperateReq,
  626. }
  627. // 铁合金
  628. global.M2A_FuncodeTopic[global.TOPIC_REQ_WAREHOUSE_PRESALE_THJ] = []int{
  629. global.SpotPresaleListingOrderReq,
  630. }
  631. global.M2A_FuncodeTopic[global.TOPIC_REQ_WAREHOUSE_PRESALE_GZ] = []int{
  632. global.GZPresaleApplyReq,
  633. global.GZPresaleOrderReq,
  634. }
  635. global.M2A_FuncodeTopic[global.TOPIC_REQ_WAREHOUSE_CENTERPURCHASE_GZ] = []int{
  636. global.GZCenterPurchaseApplyReq,
  637. global.GZCenterPurchaseOrderReq,
  638. global.BSWMSReckonPayReq,
  639. }
  640. }