package consume import ( "applet/app/cfg" "applet/app/db" db2 "applet/app/flexible_employment/db" "applet/app/flexible_employment/enum" "applet/app/flexible_employment/svc" "applet/app/flexible_employment/utils/aes" "applet/app/lib/flexible_employment" "applet/app/utils" "applet/app/utils/logx" "applet/consume/md" "code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit" "code.fnuoos.com/go_rely_warehouse/zyos_go_order_relate_rule.git/rule/one_circles" "encoding/json" "errors" "fmt" "github.com/streadway/amqp" "time" ) func FlexibleEmploymentWithdrawForPupiaoConsume(queue md.MqQueue) { fmt.Println(">>>>>>>>>>>>FlexibleEmploymentWithdrawForPupiaoConsume>>>>>>>>>>>>") ch, err := rabbit.Cfg.Pool.GetChannel() if err != nil { logx.Error(err) return } defer ch.Release() //1、将自己绑定到交换机上 ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey) //2、取出数据进行消费 ch.Qos(1) delivery := ch.Consume(queue.Name, false) one_circles.Init(cfg.RedisAddr) var res amqp.Delivery var ok bool for { res, ok = <-delivery if ok == true { err = handleFlexibleEmploymentWithdrawForPupiaoConsume(res.Body) fmt.Println("err ::: ", err) if err != nil { fmt.Println("FlexibleEmploymentWithdrawForPupiaoConsume_ERR:::::", err.Error()) //_ = res.Reject(true) _ = res.Reject(false) var msg interface{} json.Unmarshal(res.Body, &msg) if err.Error() == "Connection timed out" { //TODO::重新推回队列末尾,避免造成队列堵塞 ch.Publish(queue.ExchangeName, msg, queue.RoutKey) } else { //TODO::推入新的队列中备份 utils.FilePutContents("FlexibleEmploymentWithdrawForPupiaoConsume_ERR", utils.SerializeStr(err.Error())) ch.Publish("zhios.app.user.withdraw.apply.exception.exchange", msg, "pupiao") } } else { err = res.Ack(true) } } else { panic(errors.New("error getting message")) } } fmt.Println("get msg done") } func handleFlexibleEmploymentWithdrawForPupiaoConsume(msgData []byte) error { var ms string err := json.Unmarshal(msgData, &ms) if err != nil { return err } time.Sleep(time.Microsecond * 200) // 等待200毫秒 //1、解析mq中queue的数据结构体 var msg struct { Uid string `json:"uid"` Nickname string `json:"nickname"` MasterId string `json:"master_id"` AppName string `json:"app_name"` ApplyOrder string `json:"apply_order"` ActualAmount string `json:"actual_amount"` Oid string `json:"oid"` MobCfg interface{} `json:"mob_cfg"` } err = json.Unmarshal([]byte(ms), &msg) if err != nil { return err } fmt.Println("pupiao_message:::::::::::>>>>>>>>>") fmt.Println(msg) if db.DBs[msg.MasterId] == nil { return nil } engine := db.DBs[msg.MasterId] //1、查找对应记录 flexibleEmploymentPupiaoOrdDb := db2.FlexibleEmploymentPupiaoOrdDb{} flexibleEmploymentPupiaoOrdDb.Set(msg.MasterId) flexibleEmploymentPupiaoOrd, err := flexibleEmploymentPupiaoOrdDb.Get(msg.Oid) if err != nil { return err } if flexibleEmploymentPupiaoOrd == nil { return errors.New("未查询到对应订单记录") } flexibleEmploymentPuiaoBasicDb := db2.FlexibleEmploymentPuiaoBasicDb{} flexibleEmploymentPuiaoBasicDb.Set() basic, err := flexibleEmploymentPuiaoBasicDb.GetBasic(msg.MasterId) if err != nil { return err } //2、发起制单 puPiao := flexible_employment.NewPuPiao(basic.AppId, basic.AppSecret) var userAESData = []map[string]string{ { "outTradeNo": "o_" + utils.Int64ToStr(flexibleEmploymentPupiaoOrd.WithdrawApplyId), "settleType": flexibleEmploymentPupiaoOrd.SettleType, "payeeName": flexibleEmploymentPupiaoOrd.PayeeName, "payeeIdCard": flexibleEmploymentPupiaoOrd.PayeeIdCard, "payeePhone": flexibleEmploymentPupiaoOrd.PayeePhone, "payeeNo": flexibleEmploymentPupiaoOrd.PayeeNo, "orderAmount": flexibleEmploymentPupiaoOrd.TotalAmount, }, } str, _ := json.Marshal(userAESData) itemAESContent, _ := aes.AesEncryptByECB([]byte(basic.AppSecret), string(str)) result, err := puPiao.Curl(enum.OpenApiPaymentReceiveOrder, map[string]interface{}{ "outBatchNo": flexibleEmploymentPupiaoOrd.OutBatchNo, "hrcompanyId": basic.HrCompanyId, "settleAccountId": basic.SettleAccountId, "totalCount": "1", "totalAmount": flexibleEmploymentPupiaoOrd.TotalAmount, "itemAESContent": itemAESContent, }) if err != nil { return err } var response struct { IsSuccess string `json:"isSuccess"` Charset string `json:"charset"` ErrorCode string `json:"errorCode"` ErrorMsg string `json:"errorMsg"` Data struct { PlatformBatchNo string `json:"platformBatchNo"` OutBatchNo string `json:"outBatchNo"` } `json:"data"` } if err = json.Unmarshal(utils.Serialize(result), &response); err != nil { return err } if response.IsSuccess == "F" { flexibleEmploymentPupiaoOrd.BatchStatus = 1 updateAck, err := flexibleEmploymentPupiaoOrdDb.Update(flexibleEmploymentPupiaoOrd.Id, flexibleEmploymentPupiaoOrd, "batch_status") if err != nil { return err } if updateAck <= 0 { return errors.New("更新 flexible_employment_pupiao_ord 状态失败") } //TODO::制单失败,将处理提现失败状态 finWithdrawApply, err := db.UserWithDrawApplyByUIDById(engine, utils.Int64ToStr(flexibleEmploymentPupiaoOrd.WithdrawApplyId)) if err != nil { return err } session := engine.NewSession() defer session.Close() session.Begin() err = svc.DealFailResultForPuPiao(session, finWithdrawApply, msg.MasterId, response.ErrorMsg) if err != nil { _ = session.Rollback() return err } return session.Commit() } return nil }