MTP20_ACCESS/initialize/rabbitmq.go

package initialize

import (
	"encoding/base64"
	"fmt"
	"mtp20access/global"
	rsp "mtp20access/model/mq/response"
	"mtp20access/packet"
	"mtp20access/res/pb"
	"mtp20access/utils"

	"github.com/golang/protobuf/proto"
	"github.com/streadway/amqp"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
)

func RabbitMQ() *global.RabbitMQ {
	url := global.M2A_CONFIG.Rabbitmq.Url
	connection, err := amqp.Dial(url)
	if err != nil {
		global.M2A_LOG.Error("rabbitmq connect failed, err:", zap.Error(err))
		return nil
	}
	channel, err := connection.Channel()
	if err != nil {
		global.M2A_LOG.Error("rabbitmq open channel failed, err:", zap.Error(err))
		return nil
	}
	global.M2A_LOG.Info("rabbitmq connect successed.")

	return &global.RabbitMQ{
		Connection: connection,
		Channel:    channel,
	}
}

// MQProc 消息处理对象
type MQProc struct{}

// process 消息处理接口
func (t *MQProc) Process(topic, queuename string, msg *[]byte) {
	// info := fmt.Sprintf("rabbitmq receive message from: topic[%s] queue[%s] contentLen[%d]",
	// 	topic,
	// 	queuename,
	// 	len(string(*msg)))
	// global.M2A_LOG.Info(info)

	if funcode, sessionId, bytes, serialNumber, err := t.getRspProtobuf(msg); err == nil && bytes != nil {
		// 尝试获取对应异步任务
		if client, exists := global.M2A_Clients[int(sessionId)]; exists {
			key := fmt.Sprintf("%v_%v_%v", sessionId, funcode, serialNumber)
			// 银行服务相关的回复流水号是错误的，所以需要特殊处理
			if int(funcode) == global.T2bBankSignRsp ||
				int(funcode) == global.T2bBankCancelSignRsp ||
				int(funcode) == global.T2bBankWithdrawRsp ||
				int(funcode) == global.T2bBankDepositRsp {
				key = fmt.Sprintf("%v_%v", sessionId, funcode)
			}
			asyncTask := client.GetAsyncTask(key)
			if asyncTask != nil {
				rspData := string(*bytes)
				// 判断是否要加密
				if asyncTask.IsEncrypted {
					if b, err := packet.Encrypt(*bytes, packet.AESKey, true); err != nil {
						global.M2A_LOG.Error("总线回复数据加密失败", zap.Error(err))
						return
					} else {
						rspData = base64.StdEncoding.EncodeToString(b)
					}
				}
				// 给客户端回调
				global.M2A_LOG.Info("[S->C]", zap.Any("rsp", funcode), zap.Any("data", string(rspData)))
				r := rsp.MQBodyRsp{
					FunCode:     funcode,
					IsEncrypted: asyncTask.IsEncrypted,
					Data:        rspData,
				}
				asyncTask.Rsp <- r
			} else {
				global.M2A_LOG.Info("找不到对应KEY的异步任务", zap.Any("key", key), zap.Any("AsyncMap", client.GetAllAsyncTask()))
			}
		} else {
			global.M2A_LOG.Info("找不到对应的client", zap.Any("sessionId", sessionId))
		}
	}
}

// getRspProtobuf 将总线回复的数据反序列化为Protobuf
func (t *MQProc) getRspProtobuf(msg *[]byte) (funcode uint32, sessionId uint32, bytes *[]byte, serialNumber uint32, err error) {
	// 分解总线包信息
	funcode = utils.BytesToUint32((*msg)[0:4])
	sessionId = utils.BytesToUint32((*msg)[4:8])
	b := (*msg)[8:]

	global.M2A_LOG.Info("收到总线订阅消息", zap.Any("funcode", funcode), zap.Any("sessionId", sessionId), zap.Any("len", len(b)))

	switch int(funcode) {
	case global.ModifyPwdRsp: // 修改账户密码应答
		var p pb.ModifyPwdRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.UserReceiveInfoRsp: // 新增修改收货地址请求响应
		var p pb.UserReceiveInfoRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.DelUserReceiveInfoRsp: // 删除收货地址请求响应
		var p pb.DelUserReceiveInfoRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.UserReceiptInfoRsp: // 新增修改用户发票信息请求响应
		var p pb.UserReceiptInfoRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.DelUserReceiptInfoRsp: // 删除用户发票信息请求响应
		var p pb.DelUserReceiptInfoRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.T2bBankSignRsp: // 签约应答
		var p pb.T2BBankSignRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.T2bBankCancelSignRsp: // 解约应答
		var p pb.T2BBankCancelSignRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.T2bBankWithdrawRsp: // 出金应答
		var p pb.T2BBankWithdrawRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.T2bBankDepositRsp: // 入金应答
		var p pb.T2BBankDepositRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.WarehouseApplyRsp: // 仓库申请应答
		var p pb.WarehouseApplyRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.PerformanceContractedApplyRsp: // 违约申请应答
		var p pb.PerformanceContractedApplyRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.PerformanceDelayApplyRsp: // 延期申请应答
		var p pb.PerformanceDelayApplyRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.PerformanceManualConfirmRsp: // 履约手动确认应答
		var p pb.PerformanceManualConfirmRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.PerformanceModifyContactRsp: // 履约修改联络信息回应
		var p pb.PerformanceModifyContactRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.GoodsFavoriteOperateRsp: // 商品收藏操作接口应答
		var p pb.GoodsFavoriteOperateRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.ZSBuyOrderListingRsp: // 钻石买挂牌接口应答
		var p pb.ZSBuyOrderListingRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.ZSSellOrderListingRsp: // 钻石卖挂牌接口响应
		var p pb.ZSSellOrderListingRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.ZSBuyOrderDestingRsp: // 钻石卖摘牌申请接口响应
		var p pb.ZSBuyOrderDestingRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.ZSSellOrderDestingApplyOperateRsp: // 钻石卖摘牌申请操作接口应答
		var p pb.ZSSellOrderDestingApplyOperateRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.ZSBuyOrderDestingNegPriceRsp: // 买摘牌询价接口应答
		var p pb.ZSBuyOrderDestingNegPriceRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.ZSBuyOrderDestingNegPriceOperateRsp: // 买摘牌询价操作接口应答
		var p pb.ZSBuyOrderDestingNegPriceOperateRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.WRListingCancelOrderRsp: // 挂牌撤单应答
		var p pb.WRListingCancelOrderRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.ReceiptZSOutApplyRsp: // 钻石出库申请接口响应
		var p pb.ReceiptZSOutApplyRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.SpotPresaleListingOrderRsp: // 铁合金现货预售挂牌接口应答
		var p pb.SpotPresaleListingOrderRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.CJJCMemberOperateRsp: // 出境检测会员操作响应
		var p pb.CJJCMemberOperateRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.BSFWMemberOperateRsp: // 保税服务会员操作响应
		var p pb.BSFWMemberOperateRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	case global.AccountFundInfoRsp: // 账户资金信息响应
		var p pb.AccountFundInfoRsp
		if err = proto.Unmarshal(b, &p); err != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		}
		if bs, e := protojson.Marshal(&p); e != nil {
			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
			return
		} else {
			bytes = &bs
			serialNumber = p.GetHeader().GetRequestID()
		}
	}

	return
}

// RabbitMQSubscribeTopic 订阅主题
func RabbitMQSubscribeTopic() (err error) {
	// 订阅需要的总线响应主题
	if err = global.SubscribeTopic(global.TOPIC_RSP_QKERNEL); err != nil {
		global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
		return
	}
	if err = global.SubscribeTopic(global.TOPIC_MANAGE_RSP); err != nil {
		global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
		return
	}
	if err = global.SubscribeTopic(global.TOPIC_RSP_BANK); err != nil {
		global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
		return
	}
	if err = global.SubscribeTopic(global.TOPIC_RSP_MONEY); err != nil {
		global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
		return
	}
	if err = global.SubscribeTopic(global.TOPIC_PERFORMANCE_RSP); err != nil {
		global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
		return
	}
	if err = global.SubscribeTopic(global.TOPIC_WAREHOUSE_RECIEPT_RSP); err != nil {
		global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
		return
	}
	if err = global.SubscribeTopic(global.TOPIC_RSP_WAREHOUSE_TRADE); err != nil {
		global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
		return
	}
	if err = global.SubscribeTopic(global.TOPIC_RSP_WAREHOUSE_TRADE_GZ); err != nil {
		global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
		return
	}
	if err = global.SubscribeTopic(global.TOPIC_RSP_CJBS_TRADE_GZ); err != nil {
		global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
		return
	}

	// 铁合金
	if err = global.SubscribeTopic(global.TOPIC_RSP_WAREHOUSE_PRESALE_THJ); err != nil {
		global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
		return
	}
	if err = global.SubscribeTopic(global.TOPIC_MANAGER_THJ_NTF); err != nil {
		global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
		return
	}

	global.M2A_LOG.Info("rabbitmq subscribe topic successed.")

	return
}

// StartRabbitMQReceive 开始接收总线消息
func StartRabbitMQReceive() {
	t := &MQProc{}

	go func() {
		for _, subinfo := range global.SubInfos {
			global.Receive(subinfo.Topic, subinfo.QueueName, t)
		}
	}()
}

// InitFuncodeTopic 初始化功能码主题MAP
func InitFuncodeTopic() {
	if global.M2A_FuncodeTopic == nil {
		global.M2A_FuncodeTopic = make(map[string][]int)
	}

	global.M2A_FuncodeTopic[global.TOPIC_REQ_QKERNEL] = []int{
		global.ModifyPwdReq,
	}
	global.M2A_FuncodeTopic[global.TOPIC_MANAGE_REQ] = []int{
		global.UserReceiveInfoReq,
		global.DelUserReceiveInfoReq,
		global.UserReceiveIsDefaultReq,
		global.UserReceiptInfoReq,
		global.DelUserReceiptInfoReq,
		global.WarehouseApplyReq,
	}
	global.M2A_FuncodeTopic[global.TOPIC_REQ_BANK] = []int{
		global.T2bBankSignReq,
		global.T2bBankCancelSignReq,
		global.T2bBankWithdrawReq,
		global.T2bBankDepositReq,
	}
	global.M2A_FuncodeTopic[global.TOPIC_REQ_MONEY] = []int{
		global.AccountFundInfoReq,
	}
	global.M2A_FuncodeTopic[global.TOPIC_PERFORMANCE_REQ] = []int{
		global.PerformanceContractedApplyReq,
		global.PerformanceDelayApplyReq,
		global.PerformanceManualConfirmReq,
		global.PerformanceModifyContactReq,
	}
	global.M2A_FuncodeTopic[global.TOPIC_WAREHOUSE_RECIEPT] = []int{
		global.ReceiptZSOutApplyReq,
	}
	global.M2A_FuncodeTopic[global.TOPIC_REQ_WAREHOUSE_TRADE] = []int{
		global.WRListingCancelOrderReq,
	}
	global.M2A_FuncodeTopic[global.TOPIC_REQ_WAREHOUSE_TRADE_GZ] = []int{
		global.GoodsFavoriteOperateReq,
		global.ZSBuyOrderListingReq,
		global.ZSSellOrderListingReq,
		global.ZSBuyOrderDestingReq,
		global.ZSSellOrderDestingApplyReq,
		global.ZSSellOrderDestingApplyOperateReq,
		global.ZSBuyOrderDestingNegPriceReq,
		global.ZSBuyOrderDestingNegPriceOperateReq,
	}
	global.M2A_FuncodeTopic[global.TOPIC_REQ_CJBS_TRADE_GZ] = []int{
		global.CJJCMemberOperateReq,
		global.BSFWMemberOperateReq,
	}

	// 铁合金
	global.M2A_FuncodeTopic[global.TOPIC_REQ_WAREHOUSE_PRESALE_THJ] = []int{
		global.SpotPresaleListingOrderReq,
	}
}