|
@@ -0,0 +1,174 @@ |
|
|
|
|
|
package consume |
|
|
|
|
|
|
|
|
|
|
|
import ( |
|
|
|
|
|
"applet/app/db" |
|
|
|
|
|
"applet/app/db/model" |
|
|
|
|
|
"applet/app/utils" |
|
|
|
|
|
"applet/app/utils/logx" |
|
|
|
|
|
"applet/consume/md" |
|
|
|
|
|
"code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit" |
|
|
|
|
|
"encoding/json" |
|
|
|
|
|
"errors" |
|
|
|
|
|
"fmt" |
|
|
|
|
|
"github.com/streadway/amqp" |
|
|
|
|
|
"time" |
|
|
|
|
|
"xorm.io/xorm" |
|
|
|
|
|
) |
|
|
|
|
|
|
|
|
|
|
|
func ZhiosAutoUnFreeze(queue md.MqQueue) { |
|
|
|
|
|
fmt.Println(">>>>>>>>>>>>>>>>>>>>>>>>") |
|
|
|
|
|
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) |
|
|
|
|
|
|
|
|
|
|
|
var res amqp.Delivery |
|
|
|
|
|
var ok bool |
|
|
|
|
|
for { |
|
|
|
|
|
res, ok = <-delivery |
|
|
|
|
|
if ok == true { |
|
|
|
|
|
//fmt.Println(string(res.Body)) |
|
|
|
|
|
fmt.Println(">>>>>>>>>>>>>>>>CanalOrderConsume<<<<<<<<<<<<<<<<<<<<<<<<<") |
|
|
|
|
|
err = AutoUnFreeze(res.Body) |
|
|
|
|
|
//_ = res.Reject(false) |
|
|
|
|
|
if err == nil { |
|
|
|
|
|
_ = res.Ack(true) |
|
|
|
|
|
} |
|
|
|
|
|
} else { |
|
|
|
|
|
panic(errors.New("error getting message")) |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
|
|
|
fmt.Println("get msg done") |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
func AutoUnFreeze(msg []byte) error { |
|
|
|
|
|
//1、解析canal采集至mq中queue的数据结构体 |
|
|
|
|
|
var canalMsg *md.ZhiosAutoUnFreezes |
|
|
|
|
|
fmt.Println(string(msg)) |
|
|
|
|
|
var tmpString string |
|
|
|
|
|
err := json.Unmarshal(msg, &tmpString) |
|
|
|
|
|
if err != nil { |
|
|
|
|
|
fmt.Println(err.Error()) |
|
|
|
|
|
return err |
|
|
|
|
|
} |
|
|
|
|
|
fmt.Println(tmpString) |
|
|
|
|
|
err = json.Unmarshal([]byte(tmpString), &canalMsg) |
|
|
|
|
|
if err != nil { |
|
|
|
|
|
return err |
|
|
|
|
|
} |
|
|
|
|
|
mid := canalMsg.Mid |
|
|
|
|
|
eg := db.DBs[mid] |
|
|
|
|
|
if eg == nil { |
|
|
|
|
|
return nil |
|
|
|
|
|
} |
|
|
|
|
|
tmp := db.GetUnFreezeUser(eg, canalMsg.Id) |
|
|
|
|
|
if tmp == nil { |
|
|
|
|
|
return nil |
|
|
|
|
|
} |
|
|
|
|
|
if utils.StrToFloat64(tmp.Money) == 0 { |
|
|
|
|
|
return nil |
|
|
|
|
|
} |
|
|
|
|
|
sess := eg.NewSession() |
|
|
|
|
|
defer sess.Close() |
|
|
|
|
|
sess.Begin() |
|
|
|
|
|
if tmp.CoinId == 0 { |
|
|
|
|
|
err = finDoing(sess, tmp) |
|
|
|
|
|
} else { |
|
|
|
|
|
err = coinDoing(sess, tmp) |
|
|
|
|
|
} |
|
|
|
|
|
if err != nil { |
|
|
|
|
|
sess.Rollback() |
|
|
|
|
|
return nil |
|
|
|
|
|
} |
|
|
|
|
|
sess.Commit() |
|
|
|
|
|
return nil |
|
|
|
|
|
} |
|
|
|
|
|
func finDoing(sess *xorm.Session, data *model.UnFreezeUser) error { |
|
|
|
|
|
|
|
|
|
|
|
user, _ := db.UserProfileFindByIDSess(sess, data.Uid) |
|
|
|
|
|
if user == nil { |
|
|
|
|
|
return nil |
|
|
|
|
|
} |
|
|
|
|
|
if utils.StrToFloat64(user.FinInvalid) <= 0 { |
|
|
|
|
|
return nil |
|
|
|
|
|
} |
|
|
|
|
|
money := data.Money |
|
|
|
|
|
if utils.StrToFloat64(user.FinInvalid) < utils.StrToFloat64(data.Money) { |
|
|
|
|
|
money = user.FinInvalid |
|
|
|
|
|
} |
|
|
|
|
|
beforeAmount := user.FinValid |
|
|
|
|
|
user.FinInvalid = utils.Float64ToStrByPrec(utils.StrToFloat64(user.FinInvalid)-utils.StrToFloat64(money), 8) |
|
|
|
|
|
user.FinValid = utils.Float64ToStrByPrec(utils.StrToFloat64(user.FinValid)+utils.StrToFloat64(money), 8) |
|
|
|
|
|
update, err := sess.Where("uid=?", user.Uid).Cols("fin_valid,fin_invalid").Update(user) |
|
|
|
|
|
if update == 0 || err != nil { |
|
|
|
|
|
return errors.New("失败") |
|
|
|
|
|
} |
|
|
|
|
|
var flowData = model.FinUserFlow{ |
|
|
|
|
|
Uid: user.Uid, |
|
|
|
|
|
Type: 0, |
|
|
|
|
|
Amount: money, |
|
|
|
|
|
BeforeAmount: beforeAmount, |
|
|
|
|
|
AfterAmount: user.FinValid, |
|
|
|
|
|
OrdType: "auto_freeze", |
|
|
|
|
|
OrdId: "", |
|
|
|
|
|
OrdTitle: "自动解冻金额", |
|
|
|
|
|
OrdAction: 30, |
|
|
|
|
|
OrdTime: int(time.Now().Unix()), |
|
|
|
|
|
State: 2, |
|
|
|
|
|
CreateAt: time.Now(), |
|
|
|
|
|
UpdateAt: time.Now(), |
|
|
|
|
|
} |
|
|
|
|
|
insert, err := sess.Insert(&flowData) |
|
|
|
|
|
if insert == 0 || err != nil { |
|
|
|
|
|
return errors.New("失败") |
|
|
|
|
|
} |
|
|
|
|
|
return nil |
|
|
|
|
|
} |
|
|
|
|
|
func coinDoing(sess *xorm.Session, data *model.UnFreezeUser) error { |
|
|
|
|
|
user, _ := db.UserVirtualAmountFindById(sess, data.Uid, data.CoinId) |
|
|
|
|
|
if user == nil { |
|
|
|
|
|
return nil |
|
|
|
|
|
} |
|
|
|
|
|
if utils.StrToFloat64(user.FreezeAmount) <= 0 { |
|
|
|
|
|
return nil |
|
|
|
|
|
} |
|
|
|
|
|
money := data.Money |
|
|
|
|
|
if utils.StrToFloat64(user.FreezeAmount) < utils.StrToFloat64(data.Money) { |
|
|
|
|
|
money = user.FreezeAmount |
|
|
|
|
|
} |
|
|
|
|
|
beforeAmount := user.Amount |
|
|
|
|
|
beforeFreezeAmount := user.FreezeAmount |
|
|
|
|
|
user.FreezeAmount = utils.Float64ToStrByPrec(utils.StrToFloat64(user.FreezeAmount)-utils.StrToFloat64(money), 8) |
|
|
|
|
|
user.Amount = utils.Float64ToStrByPrec(utils.StrToFloat64(user.Amount)+utils.StrToFloat64(money), 8) |
|
|
|
|
|
update, err := sess.Where("uid=?", user.Uid).Cols("freeze_amount,amount").Update(user) |
|
|
|
|
|
if update == 0 || err != nil { |
|
|
|
|
|
return errors.New("失败") |
|
|
|
|
|
} |
|
|
|
|
|
var flowData = model.UserVirtualCoinFlow{ |
|
|
|
|
|
Uid: user.Uid, |
|
|
|
|
|
CoinId: data.CoinId, |
|
|
|
|
|
Direction: 1, |
|
|
|
|
|
Title: "自动解冻", |
|
|
|
|
|
OrdId: "", |
|
|
|
|
|
Date: "", |
|
|
|
|
|
Amout: money, |
|
|
|
|
|
BeforeAmout: beforeAmount, |
|
|
|
|
|
AfterAmout: user.Amount, |
|
|
|
|
|
CreateTime: time.Now(), |
|
|
|
|
|
TransferType: 200, |
|
|
|
|
|
FreezeBeforeAmout: beforeFreezeAmount, |
|
|
|
|
|
FreezeAfterAmout: user.FreezeAmount, |
|
|
|
|
|
} |
|
|
|
|
|
insert, err := sess.Insert(&flowData) |
|
|
|
|
|
if insert == 0 || err != nil { |
|
|
|
|
|
return errors.New("失败") |
|
|
|
|
|
} |
|
|
|
|
|
return nil |
|
|
|
|
|
} |