egg_mq_consume/consume/egg_canal_user_down_consume.go

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_models.git/src/model"
	"code.fnuoos.com/EggPlanet/egg_system_rules.git"
	"code.fnuoos.com/EggPlanet/egg_system_rules.git/rule"
	"code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/streadway/amqp"
	"sort"
	"time"
)

func EggCanalUserDownConsume(queue md.MqQueue) {
	fmt.Println(">>>>>>>>>>>>EggCanalUserDownConsume>>>>>>>>>>>>")
	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 = handleEggCanalUserDownConsume(res.Body)
			err = res.Ack(true)
			fmt.Println("err ::: ", err)
		} else {
			panic(errors.New("error getting message"))
		}
	}
	fmt.Println("get msg done")
}

func handleEggCanalUserDownConsume(msgData []byte) error {
	var msg *md.CommUserId
	err := json.Unmarshal(msgData, &msg)
	if err != nil {
		return err
	}
	engine := db.Db
	NewUserDb := implement.NewUserDb(engine)
	user, _ := NewUserDb.GetUser(utils.StrToInt64(msg.Uid))
	levelDb := implement.NewUserLevelDb(engine)
	levelAll, _ := levelDb.UserLevelAllByAsc()
	level := 1
	if len(levelAll) > 0 {
		level = levelAll[0].Id
	}
	sort.Slice(levelAll, func(i, j int) bool {
		return levelAll[i].LevelWeight > levelAll[j].LevelWeight
	})
	isHas := 0
	task := make([]map[string]string, 0)
	for _, v := range levelAll {
		if v.Id == user.Level {
			isHas = 1
		}
		if isHas == 1 {
			task, err = rule.UserUpgradeTask(engine, int(user.Id), v.Id)
			if err != nil {
				continue
			}
			level = v.Id
		}
	}
	//升级
	oldLevel := user.Level
	user.Level = level
	_, err = engine.Where("id=?", user.Id).Cols("level").Update(user)
	if err != nil {
		return err
	}
	var tmp = model.UserUpgradeLevel{
		Uid:        int(user.Id),
		Level:      level,
		OldLv:      oldLevel,
		CreateTime: time.Now(),
		Task:       utils.SerializeStr(task),
	}
	engine.Insert(&tmp)
	return nil
}