add batch send message

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/im_egg_energy_batch_send_message_consume.go

@@ -0,0 +1,87 @@
+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"))
+		}
+	}
+	fmt.Println("get msg done")
+}
+
+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/init.go

@@ -20,6 +20,7 @@ func initConsumes() {
 	jobs[consumeMd.EggEnergyStartLevelDividendFunName] = EggEnergyStartLevelDividendConsume
 	jobs[consumeMd.EggEnergyDealPlatformRevenueDataFunName] = EggEnergyDealPlatformRevenueDataConsume
 	jobs[consumeMd.EggEnergyDealFundDataFunName] = EggEnergyDealFundDataConsume
+	jobs[consumeMd.IMEggEnergyBatchSendMessageDataFunName] = IMEggEnergyBatchSendMessageDataConsume
 }
 
 func Run() {

consume/md/consume_key.go

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

go.mod

@@ -7,7 +7,7 @@ replace code.fnuoos.com/EggPlanet/egg_models.git => E:/company/Egg/egg_models
 replace code.fnuoos.com/EggPlanet/egg_system_rules.git => E:/company/Egg/egg_system_rules
 
 require (
-	code.fnuoos.com/EggPlanet/egg_models.git v0.2.1-0.20241114063419-cb68a0ed34ee
+	code.fnuoos.com/EggPlanet/egg_models.git v0.2.1-0.20241119093836-37be936b83fc
 	code.fnuoos.com/EggPlanet/egg_system_rules.git v0.0.2
 	code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git v0.0.5
 	github.com/boombuler/barcode v1.0.1
@@ -21,6 +21,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
@@ -52,11 +53,8 @@ require (
 	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.19.0 // indirect
+	github.com/nxadm/tail v1.4.8 // 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
@@ -65,10 +63,11 @@ require (
 	golang.org/x/arch v0.7.0 // indirect
 	golang.org/x/crypto v0.21.0 // indirect
 	golang.org/x/lint v0.0.0-20200302205851-738671d3881b // indirect
-	golang.org/x/net v0.22.0 // indirect
+	golang.org/x/net v0.23.0 // indirect
 	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