update

app/cfg/init_cfg.go

@@ -17,6 +17,7 @@ var (
 	DB            *DBCfg
 	MQ            *MQCfg
 	Log           *LogCfg
+	ImLogicRpc    *ImLogicRpcCfg
 )
 
 // 初始化配置文件，将cfg.yml读入到内存
@@ -48,3 +49,8 @@ func InitCfg() {
 	MQ = &conf.MQ
 	RedisPassword = conf.RedisPassword
 }
+
+type ImLogicRpcCfg struct {
+	URL  string `yaml:"url"`
+	PORT string `yaml:"port"`
+}

app/utils/rpc_client.go

@@ -0,0 +1,64 @@
+package utils
+
+import (
+	"applet/pkg/pb"
+	"context"
+	"fmt"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
+	"strconv"
+	"time"
+)
+
+func GetBusinessIntClient(url, port string) pb.BusinessIntClient {
+	target := fmt.Sprintf("%s:%s", url, port)
+	conn, err := grpc.Dial(target, grpc.WithInsecure())
+	if err != nil {
+		fmt.Println(err)
+		return nil
+	}
+	return pb.NewBusinessIntClient(conn)
+}
+
+func GetBusinessExtClient(url, port string) pb.BusinessExtClient {
+	target := fmt.Sprintf("%s:%s", url, port)
+	conn, err := grpc.Dial(target, grpc.WithInsecure())
+	//defer conn.Close()
+	if err != nil {
+		fmt.Println(err)
+		return nil
+	}
+	return pb.NewBusinessExtClient(conn)
+}
+
+func GetLogicExtClient(url, port string) pb.LogicExtClient {
+	target := fmt.Sprintf("%s:%s", url, port)
+	conn, err := grpc.Dial(target, grpc.WithInsecure())
+	if err != nil {
+		fmt.Println(err)
+		return nil
+	}
+	return pb.NewLogicExtClient(conn)
+}
+
+func GetCtx(token, userId, deviceId, masterId string) context.Context {
+	if userId == "" {
+		userId = "1"
+	}
+	if deviceId == "" {
+		deviceId = "1"
+	}
+	if token == "" {
+		token = "0"
+	}
+	return metadata.NewOutgoingContext(context.TODO(), metadata.Pairs(
+		"user_id", userId,
+		"device_id", deviceId,
+		"token", token,
+		"master_id", masterId,
+		"request_id", strconv.FormatInt(time.Now().UnixNano(), 10)))
+}
+
+func UnixMilliTime(t time.Time) int64 {
+	return t.UnixNano() / 1000000
+}

consume/egg_energy_deal_user_virtual_coin_consume.go

@@ -0,0 +1,82 @@
+package consume
+
+import (
+	"applet/app/cfg"
+	"applet/app/db"
+	utils2 "applet/app/utils"
+	"applet/app/utils/logx"
+	"applet/consume/md"
+	"code.fnuoos.com/EggPlanet/egg_system_rules.git"
+	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"
+	"time"
+)
+
+func EggEnergyDealUserVirtualCoinDataConsume(queue md.MqQueue) {
+	fmt.Println(">>>>>>>>>>>>EggEnergyDealUserVirtualCoinDataConsume>>>>>>>>>>>>")
+	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 = handleEggEnergyDealUserVirtualCoinDataConsume(res.Body)
+			if err != nil {
+				fmt.Println("EggEnergyDealUserVirtualCoinDataConsume_ERR:::::", err.Error())
+				utils2.FilePutContents("EggEnergyDealUserVirtualCoinDataConsume_ERR", utils2.SerializeStr(map[string]interface{}{
+					"body": res.Body,
+					"err":  err.Error(),
+				}))
+			}
+			//_ = res.Reject(false)
+			err = res.Ack(true)
+			fmt.Println("err ::: ", err)
+		} else {
+			panic(errors.New("error getting message"))
+		}
+	}
+}
+
+func handleEggEnergyDealUserVirtualCoinDataConsume(msgData []byte) error {
+	time.Sleep(time.Duration(100) * time.Millisecond) //休眠100毫秒
+	// 1.解析mq中queue的数据结构体
+	var msg *md2.EggEnergyStructForDealUserVirtualCoinData
+	err := json.Unmarshal(msgData, &msg)
+	if err != nil {
+		return err
+	}
+
+	engine := db.Db
+	session := engine.NewSession()
+	err = rule.DealUserVirtualCoin(session, md3.DealUserVirtualCoinReq{
+		Kind:         msg.Kind,
+		Title:        msg.Title,
+		TransferType: msg.TransferType,
+		CoinId:       msg.CoinId,
+		Uid:          msg.Uid,
+		Amount:       msg.Amount,
+	})
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

consume/egg_energy_fund_data_consume.go

@@ -23,7 +23,7 @@ import (
 )
 
 func EggEnergyDealFundDataConsume(queue md.MqQueue) {
-	fmt.Println(">>>>>>>>>>>>EggEnergyStartLevelDividendConsume>>>>>>>>>>>>")
+	fmt.Println(">>>>>>>>>>>>EggEnergyDealFundDataConsume>>>>>>>>>>>>")
 	ch, err := rabbit.Cfg.Pool.GetChannel()
 	if err != nil {
 		logx.Error(err)
@@ -42,7 +42,7 @@ func EggEnergyDealFundDataConsume(queue md.MqQueue) {
 	for {
 		res, ok = <-delivery
 		if ok == true {
-			err = handleEggEnergyStartLevelDividendConsume(res.Body)
+			err = handleEggEnergyDealFundDataConsume(res.Body)
 			if err != nil {
 				fmt.Println("EggEnergyDealFundDataConsume_ERR:::::", err.Error())
 				utils2.FilePutContents("EggEnergyDealFundDataConsume_ERR", utils2.SerializeStr(map[string]interface{}{
@@ -57,7 +57,6 @@ func EggEnergyDealFundDataConsume(queue md.MqQueue) {
 			panic(errors.New("error getting message"))
 		}
 	}
-	fmt.Println("get msg done")
 }
 
 func handleEggEnergyDealFundDataConsume(msgData []byte) error {

consume/egg_energy_platform_revenue_data.go

@@ -23,7 +23,7 @@ import (
 )
 
 func EggEnergyDealPlatformRevenueDataConsume(queue md.MqQueue) {
-	fmt.Println(">>>>>>>>>>>>EggEnergyStartLevelDividendConsume>>>>>>>>>>>>")
+	fmt.Println(">>>>>>>>>>>>EggEnergyDealPlatformRevenueDataConsume>>>>>>>>>>>>")
 	ch, err := rabbit.Cfg.Pool.GetChannel()
 	if err != nil {
 		logx.Error(err)
@@ -57,7 +57,6 @@ func EggEnergyDealPlatformRevenueDataConsume(queue md.MqQueue) {
 			panic(errors.New("error getting message"))
 		}
 	}
-	fmt.Println("get msg done")
 }
 
 func handleEggEnergyDealPlatformRevenueDataConsume(msgData []byte) error {

consume/im_egg_energy_batch_send_message_consume.go

@@ -0,0 +1,86 @@
+package consume
+
+import (
+	"applet/app/cfg"
+	utils2 "applet/app/utils"
+	"applet/app/utils/logx"
+	"applet/consume/md"
+	"applet/pkg/pb"
+	"code.fnuoos.com/EggPlanet/egg_system_rules.git"
+	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"
+	"google.golang.org/protobuf/proto"
+	"time"
+)
+
+func IMEggEnergyBatchSendMessageDataConsume(queue md.MqQueue) {
+	fmt.Println(">>>>>>>>>>>>IMEggEnergyBatchSendMessageDataConsume>>>>>>>>>>>>")
+	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 = handleIMEggEnergyBatchSendMessageDataConsume(res.Body)
+			if err != nil {
+				fmt.Println("IMEggEnergyBatchSendMessageDataConsume_ERR:::::", err.Error())
+				utils2.FilePutContents("IMEggEnergyBatchSendMessageDataConsume_ERR", utils2.SerializeStr(map[string]interface{}{
+					"body": res.Body,
+					"err":  err.Error(),
+				}))
+			}
+			//_ = res.Reject(false)
+			err = res.Ack(true)
+			fmt.Println("err ::: ", err)
+		} else {
+			panic(errors.New("error getting message"))
+		}
+	}
+}
+
+func handleIMEggEnergyBatchSendMessageDataConsume(msgData []byte) error {
+	time.Sleep(time.Duration(100) * time.Millisecond) //休眠100毫秒
+	// 1.解析mq中queue的数据结构体
+	var msg *md2.IMEggEnergyStructForBatchSendMessageData
+	err := json.Unmarshal(msgData, &msg)
+	if err != nil {
+		return err
+	}
+
+	//TODO::调用im GRPC 接口
+	buf, err := proto.Marshal(&pb.Text{
+		Text: msg.Content,
+	})
+
+	_, err = utils2.GetLogicExtClient(cfg.ImLogicRpc.URL, cfg.ImLogicRpc.PORT).SendMessage(utils2.GetCtx("", utils2.Int64ToStr(msg.SendId), "", ""), &pb.SendMessageReq{
+		ReceiverType:       pb.ReceiverType(msg.ReceiverType),
+		ReceiverId:         msg.ReceiveIMId,
+		ToUserIds:          nil,
+		MessageType:        pb.MessageType(msg.MessageType),
+		MessageContent:     buf,
+		SendTime:           utils2.UnixMilliTime(time.Now()),
+		IsPersist:          true,
+		MessageContentBack: "",
+	})
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

consume/im_egg_energy_del_friend_circle_comment_consume.go

@@ -0,0 +1,77 @@
+package consume
+
+import (
+	"applet/app/cfg"
+	utils2 "applet/app/utils"
+	"applet/app/utils/cache"
+	"applet/app/utils/logx"
+	"applet/consume/md"
+	"code.fnuoos.com/EggPlanet/egg_system_rules.git"
+	md3 "code.fnuoos.com/EggPlanet/egg_system_rules.git/md"
+	md2 "code.fnuoos.com/EggPlanet/egg_system_rules.git/rule/egg_energy/md"
+	"code.fnuoos.com/go_rely_warehouse/zyos_go_es.git/es"
+	"code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"github.com/streadway/amqp"
+	"time"
+)
+
+func IMEggEnergyDelFriendCircleDataConsume(queue md.MqQueue) {
+	fmt.Println(">>>>>>>>>>>>IMEggEnergyDelFriendCircleDataConsume>>>>>>>>>>>>")
+	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 = handleIMEggEnergyDelFriendCircleCommentDataConsume(res.Body)
+			if err != nil {
+				fmt.Println("IMEggEnergyDelFriendCircleDataConsume_ERR:::::", err.Error())
+				utils2.FilePutContents("IMEggEnergyDelFriendCircleDataConsume_ERR", utils2.SerializeStr(map[string]interface{}{
+					"body": res.Body,
+					"err":  err.Error(),
+				}))
+			}
+			//_ = res.Reject(false)
+			err = res.Ack(true)
+			fmt.Println("err ::: ", err)
+		} else {
+			panic(errors.New("error getting message"))
+		}
+	}
+}
+
+func handleIMEggEnergyDelFriendCircleCommentDataConsume(msgData []byte) error {
+	time.Sleep(time.Duration(100) * time.Millisecond) //休眠100毫秒
+	// 1.解析mq中queue的数据结构体
+	var msg *md2.IMEggEnergyStructForDelFriendCircleCommentData
+	err := json.Unmarshal(msgData, &msg)
+	if err != nil {
+		return err
+	}
+
+	//1、删除es数据
+	_, err = es.DeleteDoc(md3.EggFriendCircleCommentEsAlias, msg.CommentIndexId)
+	if err != nil {
+		return err
+	}
+
+	//2、删除redis数据
+	key := fmt.Sprintf(md3.CommentLikeCacheKey, msg.CommentIndexId)
+	cache.Del(key)
+	return nil
+}

consume/init.go

@@ -18,8 +18,11 @@ func Init() {
 // 增加消费任务队列
 func initConsumes() {
 	jobs[consumeMd.EggEnergyStartLevelDividendFunName] = EggEnergyStartLevelDividendConsume
-	jobs[consumeMd.EggEnergyDealFundDataFunName] = EggEnergyDealFundDataConsume
 	jobs[consumeMd.EggEnergyDealPlatformRevenueDataFunName] = EggEnergyDealPlatformRevenueDataConsume
+	jobs[consumeMd.EggEnergyDealFundDataFunName] = EggEnergyDealFundDataConsume
+	jobs[consumeMd.EggEnergyDealUserVirtualCoinDataFunName] = EggEnergyDealUserVirtualCoinDataConsume
+	jobs[consumeMd.IMEggEnergyBatchSendMessageDataFunName] = IMEggEnergyBatchSendMessageDataConsume
+	jobs[consumeMd.IMEggEnergyDelFriendCircleDataFunName] = IMEggEnergyDelFriendCircleDataConsume
 }
 
 func Run() {

consume/md/consume_key.go

@@ -38,10 +38,40 @@ var RabbitMqQueueKeyList = []*MqQueue{
 		BindKey:        "",
 		ConsumeFunName: "EggEnergyDealPlatformRevenueDataConsume",
 	},
+	{
+		ExchangeName:   "egg.energy",
+		Name:           "egg_deal_user_virtual_coin_queue",
+		Type:           DirectQueueType,
+		IsPersistent:   false,
+		RoutKey:        "deal_user_virtual_coin",
+		BindKey:        "",
+		ConsumeFunName: "EggEnergyDealUserVirtualCoinDataConsume ",
+	},
+	{
+		ExchangeName:   "im.egg.energy",
+		Name:           "im_egg_energy_batch_send_message_queue",
+		Type:           DirectQueueType,
+		IsPersistent:   false,
+		RoutKey:        "batch_send_message_data",
+		BindKey:        "",
+		ConsumeFunName: "IMEggEnergyBatchSendMessageDataConsume",
+	},
+	{
+		ExchangeName:   "im.egg.energy",
+		Name:           "im_egg_energy_del_friend_comment_circle_queue",
+		Type:           DirectQueueType,
+		IsPersistent:   false,
+		RoutKey:        "del_friend_circle_comment_data",
+		BindKey:        "",
+		ConsumeFunName: "IMEggEnergyDelFriendCircleDataConsume",
+	},
 }
 
 const (
 	EggEnergyStartLevelDividendFunName      = "EggEnergyStartLevelDividendConsume"
 	EggEnergyDealFundDataFunName            = "EggEnergyDealFundDataConsume"
 	EggEnergyDealPlatformRevenueDataFunName = "EggEnergyDealPlatformRevenueDataConsume"
+	EggEnergyDealUserVirtualCoinDataFunName = "EggEnergyDealUserVirtualCoinDataConsume"
+	IMEggEnergyBatchSendMessageDataFunName  = "IMEggEnergyBatchSendMessageDataConsume"
+	IMEggEnergyDelFriendCircleDataFunName   = "IMEggEnergyDelFriendCircleDataConsume"
 )

go.mod

@@ -1,8 +1,6 @@
 module applet
 
-go 1.21
-
-toolchain go1.23.2
+go 1.19
 
 replace code.fnuoos.com/EggPlanet/egg_models.git => E:/company/Egg/egg_models
 
@@ -11,6 +9,7 @@ replace code.fnuoos.com/EggPlanet/egg_system_rules.git => E:/company/Egg/egg_sys
 require (
 	code.fnuoos.com/EggPlanet/egg_models.git v0.2.1-0.20241128102555-fc839292d728
 	code.fnuoos.com/EggPlanet/egg_system_rules.git v0.0.2
+	code.fnuoos.com/go_rely_warehouse/zyos_go_es.git v1.0.1-0.20241118083738-0f22da9ba0be
 	code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git v0.0.5
 	github.com/boombuler/barcode v1.0.1
 	github.com/forgoer/openssl v1.2.1
@@ -23,6 +22,7 @@ require (
 	github.com/gomodule/redigo v2.0.0+incompatible
 	github.com/makiuchi-d/gozxing v0.1.1
 	github.com/qiniu/api.v7/v7 v7.8.2
+	github.com/shopspring/decimal v1.3.1
 	github.com/sony/sonyflake v1.0.0
 	github.com/streadway/amqp v1.0.0
 	github.com/syyongx/php2go v0.9.8
@@ -45,20 +45,20 @@ require (
 	github.com/goccy/go-json v0.10.2 // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
-	github.com/google/go-cmp v0.5.9 // indirect
 	github.com/gookit/color v1.3.6 // indirect
+	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.7 // indirect
 	github.com/kr/pretty v0.3.1 // indirect
 	github.com/leodido/go-urn v1.4.0 // indirect
+	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
-	github.com/onsi/ginkgo v1.16.5 // indirect
-	github.com/onsi/gomega v1.27.8 // indirect
+	github.com/nxadm/tail v1.4.8 // indirect
+	github.com/olivere/elastic/v7 v7.0.32 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.0 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
-	github.com/shopspring/decimal v1.3.1 // indirect
 	github.com/syndtr/goleveldb v1.0.0 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	github.com/ugorji/go/codec v1.2.12 // indirect
@@ -71,6 +71,7 @@ require (
 	golang.org/x/sync v0.6.0 // indirect
 	golang.org/x/sys v0.22.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
+	golang.org/x/tools v0.9.1 // indirect
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 	google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013 // indirect
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect