package consume import ( "applet/app/cfg" "applet/app/db" "applet/app/utils" "applet/app/utils/logx" "applet/consume/md" "code.fnuoos.com/EggPlanet/egg_models.git/src/implement" "code.fnuoos.com/EggPlanet/egg_system_rules.git" "code.fnuoos.com/EggPlanet/egg_system_rules.git/enum" md3 "code.fnuoos.com/EggPlanet/egg_system_rules.git/md" "code.fnuoos.com/EggPlanet/egg_system_rules.git/rule" md2 "code.fnuoos.com/EggPlanet/egg_system_rules.git/rule/egg_energy/md" "code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit" "encoding/json" "errors" "fmt" "github.com/streadway/amqp" ) func AddPublicPlatoonUserRelationCommissionConsume(queue md.MqQueue) { fmt.Println(">>>>>>>>>>>>AddPublicPlatoonUserRelationCommissionConsume>>>>>>>>>>>>") 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) egg_system_rules.Init(cfg.RedisAddr) var res amqp.Delivery var ok bool for { res, ok = <-delivery if ok == true { err = handleAddPublicPlatoonUserRelationCommissionConsume(res.Body) if err != nil { fmt.Println("err ::: ", err) utils.FilePutContents("AddPublicPlatoonUserRelationCommissionConsume_ERR", "[err]:"+err.Error()) _ = res.Reject(false) //TODO::重新推回队列末尾,避免造成队列堵塞 var msg *md2.AddPublicPlatoonUserRelationCommissionReq json.Unmarshal(res.Body, &msg) ch.Publish(queue.ExchangeName, msg, queue.RoutKey) } else { _ = res.Ack(true) } } else { panic(errors.New("error getting message")) } } fmt.Println("get msg done") } func handleAddPublicPlatoonUserRelationCommissionConsume(msgData []byte) error { //1、解析mq中queue的数据结构体 var msg *md2.AddPublicPlatoonUserRelationCommissionReq err := json.Unmarshal(msgData, &msg) if err != nil { return err } //2、给到直推奖励 eggEnergyBasicSettingDb := implement.NewEggEnergyBasicSettingDb(db.Db) eggEnergyBasicSetting, err := eggEnergyBasicSettingDb.EggEnergyBasicSettingGetOneByParams(map[string]interface{}{ "key": "is_open", "value": 1, }) if err != nil { return err } if eggEnergyBasicSetting != nil { var newUserRewardRules *md2.NewUserRewardRules err = json.Unmarshal([]byte(eggEnergyBasicSetting.NewUserIncentiveRules), &newUserRewardRules) if err != nil { fmt.Println("handleAddPublicPlatoonUserRelationCommissionConsume___ERR>>>>", err.Error()) } rewardValue := newUserRewardRules.InviteUserRewardValue session := db.Db.NewSession() err = rule.DealUserVirtualCoin(session, md3.DealUserVirtualCoinReq{ Kind: "add", Title: enum.UserVirtualAmountFlowTransferType.String(enum.EggEnergyRecruitNewUsersReward), TransferType: int(enum.EggEnergyRecruitNewUsersReward), CoinId: eggEnergyBasicSetting.PersonEggPointsCoinId, Uid: utils.StrToInt64(msg.RecommendUid), Amount: float64(rewardValue), }) if err != nil { return err } session.Commit() } //3、加入公排 var req []*md2.AddPublicPlatoonUserRelationCommissionReq req = append(req, msg) _, err = rule.AddPublicPlatoonUserRelationCommission(db.Db, req) fmt.Println("err::::", err) if err != nil { return err } return nil }